diff --git a/affinity/src/main/c/Makefile b/affinity/src/main/c/Makefile index 6bb98b2a4..d45f02e3c 100755 --- a/affinity/src/main/c/Makefile +++ b/affinity/src/main/c/Makefile @@ -20,12 +20,12 @@ JNI_JAVA_SOURCES := $(patsubst %,$(TARGET_DIR)/%.class,$(subst .,/,$(JAVA_CLASSE JAVA_BUILD_DIR := $(TARGET_DIR) JAVA_HOME ?= /usr/java/default -#JAVA_HOME :=/Library/Java/JavaVirtualMachines/jdk1.8.0_191.jdk/Contents/Home +#JAVA_HOME :=/Library/Java/JavaVirtualMachines/jdk1.8.0_231.jdk/Contents/Home JAVA_LIB := $(JAVA_HOME)/jre/lib JVM_SHARED_LIBS := -L$(JAVA_LIB)/amd64/server -L$(JAVA_LIB)/i386/server -L$(JAVA_LIB)/amd64/jrockit/ -L$(JAVA_LIB)/i386/jrockit/ -L$(JAVA_LIB)/ppc64le/server -L$(JAVA_LIB)/ppc64le/jrockit/ CXX=g++ -INCLUDES := -I $(JAVA_HOME)/include -I $(JAVA_HOME)/include/linux -I $(WORKING_DIR) +INCLUDES := -I $(JAVA_HOME)/include -I $(JAVA_HOME)/include/linux -I $(WORKING_DIR) -I $(JAVA_HOME)/include/darwin # classpath for javah ifdef CLASSPATH diff --git a/application/src/main/java/benchmark/datagenerator/DataGenerator.java b/application/src/main/java/benchmark/datagenerator/DataGenerator.java index 76590e86b..27aa86fd1 100644 --- a/application/src/main/java/benchmark/datagenerator/DataGenerator.java +++ b/application/src/main/java/benchmark/datagenerator/DataGenerator.java @@ -6,6 +6,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.List; + /** * Data generator for benchmarks, this class contains all common methods and attributes that can be used in each application */ @@ -45,5 +47,8 @@ public void clearDataStructures() { } /* Switch configuration, used in the dynamic data generator*/ + public List getTranToDecisionConf() { + return null; + } } diff --git a/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGenerator.java b/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGenerator.java index e619e35a3..9a255c939 100644 --- a/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGenerator.java +++ b/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGenerator.java @@ -39,6 +39,7 @@ public class GSTPGDataGenerator extends DataGenerator { private final int State_Access_Skewness; // ratio of state access, following zipf distribution private final int Ratio_of_Transaction_Aborts; // ratio of transaction aborts, fail the transaction or not. i.e. transfer amount might be invalid. private final int Ratio_of_Overlapped_Keys; // ratio of overlapped keys in transactions, which affects the dependencies and circulars. + private final int Ratio_of_Multiple_State_Access_Ratio;//ratio of multiple state access per transaction private final int Transaction_Length; // control the number of txns overlap with each other. private final ArrayList generatedKeys = new ArrayList<>(); @@ -66,6 +67,7 @@ public GSTPGDataGenerator(GSTPGDataGeneratorConfig dataConfig) { Ratio_of_Transaction_Aborts = dataConfig.Ratio_of_Transaction_Aborts; Ratio_of_Overlapped_Keys = dataConfig.Ratio_of_Overlapped_Keys; Transaction_Length = dataConfig.Transaction_Length; + Ratio_of_Multiple_State_Access_Ratio = dataConfig.Ratio_of_Multiple_State_Access_Ratio; int nKeyState = dataConfig.getnKeyStates(); @@ -104,6 +106,12 @@ protected void generateTuple() { } private GSEvent randomEvent() { + int NUM_ACCESS; + if (random.nextInt(100) < Ratio_of_Multiple_State_Access_Ratio) { + NUM_ACCESS = this.NUM_ACCESS; + } else { + NUM_ACCESS = 1; + } int[] keys = new int[NUM_ACCESS*Transaction_Length]; int writeLevel = -1; if (!isUnique) { diff --git a/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGeneratorConfig.java b/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGeneratorConfig.java index d716a0dde..4fcf75484 100644 --- a/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGeneratorConfig.java +++ b/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDataGeneratorConfig.java @@ -16,6 +16,7 @@ public class GSTPGDataGeneratorConfig extends DataGeneratorConfig { public int Ratio_of_Overlapped_Keys; public int Ratio_of_Transaction_Aborts; public int Transaction_Length; + public int Ratio_of_Multiple_State_Access_Ratio; @Override public void initialize(Configuration config) { @@ -25,5 +26,6 @@ public void initialize(Configuration config) { Ratio_of_Overlapped_Keys = config.getInt("Ratio_of_Overlapped_Keys", 0); Ratio_of_Transaction_Aborts = config.getInt("Ratio_of_Transaction_Aborts", 0); Transaction_Length = config.getInt("Transaction_Length", 1); + Ratio_of_Multiple_State_Access_Ratio = config.getInt("Ratio_of_Multiple_State_Access_Ratio",100); } } \ No newline at end of file diff --git a/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDynamicDataGenerator.java b/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDynamicDataGenerator.java index 91c08cbfa..f206247c8 100644 --- a/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDynamicDataGenerator.java +++ b/application/src/main/java/benchmark/datagenerator/apps/GS/TPGTxnGenerator/GSTPGDynamicDataGenerator.java @@ -26,7 +26,9 @@ public class GSTPGDynamicDataGenerator extends DynamicWorkloadGenerator { private int Ratio_of_Transaction_Aborts; // ratio of transaction aborts, fail the transaction or not. i.e. transfer amount might be invalid. private int Ratio_of_Overlapped_Keys; // ratio of overlapped keys in transactions, which affects the dependencies and circulars. private int Transaction_Length; + private int Ratio_of_Multiple_State_Access_Ratio;//ratio of multiple state access per transaction private int tthread; + private int nKeyState; // control the number of txns overlap with each other. private ArrayList generatedKeys = new ArrayList<>(); // independent transactions. @@ -49,21 +51,78 @@ public GSTPGDynamicDataGenerator(DynamicDataGeneratorConfig dynamicDataConfig) { } @Override - public void switchConfiguration() { - State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; - NUM_ACCESS = dynamicDataConfig.NUM_ACCESS; - Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; - Ratio_of_Overlapped_Keys = dynamicDataConfig.Ratio_of_Overlapped_Keys; - Transaction_Length = dynamicDataConfig.Transaction_Length; - - int nKeyState = dynamicDataConfig.getnKeyStates(); - int MAX_LEVEL = 256; - for (int i = 0; i < nKeyState; i++) { - idToLevel.put(i, i% MAX_LEVEL); + public void tranToDecisionConf() { + //TD,LD,PD,VDD,R_of_A,isCD,isCC, + StringBuilder stringBuilder = new StringBuilder(); + //TODO:hard code, function not sure + double td = Transaction_Length * dynamicDataConfig.getCheckpoint_interval(); + td = td *((double) Ratio_of_Overlapped_Keys/100); + stringBuilder.append(td); + stringBuilder.append(","); + double ld = Transaction_Length * dynamicDataConfig.getCheckpoint_interval(); + stringBuilder.append(ld); + stringBuilder.append(","); + double pd = Transaction_Length * dynamicDataConfig.getCheckpoint_interval() * ((double) Ratio_of_Overlapped_Keys/100) * NUM_ACCESS; + stringBuilder.append(pd); + stringBuilder.append(","); + stringBuilder.append((double) State_Access_Skewness/100); + stringBuilder.append(","); + stringBuilder.append((double) Ratio_of_Transaction_Aborts/10000); + stringBuilder.append(","); + if (AppConfig.isCyclic) { + stringBuilder.append("1,"); + } else { + stringBuilder.append("0,"); + } + if (AppConfig.complexity < 40000){ + stringBuilder.append("0,"); + } else { + stringBuilder.append("1,"); + } + stringBuilder.append(eventID+dynamicDataConfig.getShiftRate()*dynamicDataConfig.getCheckpoint_interval()*dynamicDataConfig.getTotalThreads()); + this.tranToDecisionConf.add(stringBuilder.toString()); + } + + @Override + public void switchConfiguration(String type) { + switch (type) { + case "default" : + State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; + NUM_ACCESS = dynamicDataConfig.NUM_ACCESS; + Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; + Ratio_of_Overlapped_Keys = dynamicDataConfig.Ratio_of_Overlapped_Keys; + Transaction_Length = dynamicDataConfig.Transaction_Length; + Ratio_of_Multiple_State_Access_Ratio = dynamicDataConfig.Ratio_of_Multiple_State_Access_Ratio; + + nKeyState = dynamicDataConfig.getnKeyStates(); + int MAX_LEVEL = 256; + for (int i = 0; i < nKeyState; i++) { + idToLevel.put(i, i% MAX_LEVEL); + } + keyZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); + configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); + p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + break; + case "LD" : + Transaction_Length = dynamicDataConfig.Transaction_Length; + break; + case "isCyclic" : + Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; + State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; + keyZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); + configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); + p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + AppConfig.isCyclic = true; + break; + case "complexity" : + AppConfig.complexity = 80000; + break; + case "unchanging" : + break; + default: + throw new IllegalStateException("Unexpected value: " + type); } - keyZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); - configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); - p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + tranToDecisionConf(); } @Override diff --git a/application/src/main/java/benchmark/datagenerator/apps/OB/OBTPGDynamicDataGenerator.java b/application/src/main/java/benchmark/datagenerator/apps/OB/OBTPGDynamicDataGenerator.java index e510cfc7e..17e645aab 100644 --- a/application/src/main/java/benchmark/datagenerator/apps/OB/OBTPGDynamicDataGenerator.java +++ b/application/src/main/java/benchmark/datagenerator/apps/OB/OBTPGDynamicDataGenerator.java @@ -33,6 +33,7 @@ public class OBTPGDynamicDataGenerator extends DynamicWorkloadGenerator { private int Transaction_Length; private int tthread; private int Ratio_of_Buying; + private int nKeyState; // control the number of txns overlap with each other. private ArrayList generatedKeys = new ArrayList<>(); // independent transactions. @@ -67,24 +68,77 @@ protected void generateTuple() { } events.add(event); } + @Override + public void tranToDecisionConf() { + //TD,LD,PD,VDD,R_of_A,isCD,isCC, + StringBuilder stringBuilder = new StringBuilder(); + //TODO:hard code, function not sure + double td = Transaction_Length * dynamicDataConfig.getCheckpoint_interval() * (1 - (double)Ratio_of_Buying/100) + + ((double)Ratio_of_Buying/100) * 1 * dynamicDataConfig.getCheckpoint_interval(); + td = td *((double) Ratio_of_Overlapped_Keys/100); + stringBuilder.append(td); + stringBuilder.append(","); + double ld = Transaction_Length * dynamicDataConfig.getCheckpoint_interval() * (1 - (double)Ratio_of_Buying/100) + + ((double)Ratio_of_Buying/100) * 1 * dynamicDataConfig.getCheckpoint_interval(); + stringBuilder.append(ld); + stringBuilder.append(","); + double pd = 0; + stringBuilder.append(pd); + stringBuilder.append(","); + stringBuilder.append((double) State_Access_Skewness/100); + stringBuilder.append(","); + stringBuilder.append((double) Ratio_of_Transaction_Aborts/10000); + stringBuilder.append(","); + if (AppConfig.isCyclic) { + stringBuilder.append("1,"); + } else { + stringBuilder.append("0,"); + } + if (AppConfig.complexity < 40000){ + stringBuilder.append("0,"); + } else { + stringBuilder.append("1,"); + } + stringBuilder.append(eventID+dynamicDataConfig.getShiftRate()*dynamicDataConfig.getCheckpoint_interval()*dynamicDataConfig.getTotalThreads()); + this.tranToDecisionConf.add(stringBuilder.toString()); + } @Override - public void switchConfiguration() { - State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; - NUM_ACCESS = dynamicDataConfig.NUM_ACCESS; - Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; - Ratio_of_Overlapped_Keys = dynamicDataConfig.Ratio_of_Overlapped_Keys; - Transaction_Length = dynamicDataConfig.Transaction_Length; - Ratio_of_Buying=dynamicDataConfig.Ratio_Of_Buying; + public void switchConfiguration(String type) { + switch (type){ + case "default" : + State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; + NUM_ACCESS = 1; + Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; + Ratio_of_Overlapped_Keys = dynamicDataConfig.Ratio_of_Overlapped_Keys; + Transaction_Length = dynamicDataConfig.Transaction_Length; + Ratio_of_Buying=dynamicDataConfig.Ratio_Of_Buying; - int nKeyState = dynamicDataConfig.getnKeyStates(); - int MAX_LEVEL = 256; - for (int i = 0; i < nKeyState; i++) { - idToLevel.put(i, i% MAX_LEVEL); + nKeyState = dynamicDataConfig.getnKeyStates(); + int MAX_LEVEL = 256; + for (int i = 0; i < nKeyState; i++) { + idToLevel.put(i, i% MAX_LEVEL); + } + keyZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); + configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); + p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + break; + case "skew" : + State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; + nKeyState = dynamicDataConfig.getnKeyStates(); + keyZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); + configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); + p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + break; + case "abort" : + Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; + break; + case "unchanging" : + break; + default: + throw new IllegalStateException("Unexpected value: " + type); } - keyZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); - configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); - p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + tranToDecisionConf(); } @Override diff --git a/application/src/main/java/benchmark/datagenerator/apps/SL/TPGTxnGenerator/SLTPGDynamicDataGenerator.java b/application/src/main/java/benchmark/datagenerator/apps/SL/TPGTxnGenerator/SLTPGDynamicDataGenerator.java index 34853e996..f61beb1fc 100644 --- a/application/src/main/java/benchmark/datagenerator/apps/SL/TPGTxnGenerator/SLTPGDynamicDataGenerator.java +++ b/application/src/main/java/benchmark/datagenerator/apps/SL/TPGTxnGenerator/SLTPGDynamicDataGenerator.java @@ -28,6 +28,7 @@ public class SLTPGDynamicDataGenerator extends DynamicWorkloadGenerator { private int Transaction_Length; // transaction length, 4 or 8 or longer private int Ratio_of_Transaction_Aborts; // ratio of transaction aborts, fail the transaction or not. i.e. transfer amount might be invalid. private int Ratio_of_Overlapped_Keys; // ratio of overlapped keys in transactions, which affects the dependencies and circulars. + private int nKeyState; // control the number of txns overlap with each other. private ArrayList generatedAcc = new ArrayList<>(); private ArrayList generatedAst = new ArrayList<>(); @@ -52,28 +53,81 @@ public SLTPGDynamicDataGenerator(DynamicDataGeneratorConfig dynamicDataConfig){ super(dynamicDataConfig); events = new ArrayList<>(nTuples); } + @Override - public void switchConfiguration() { - Ratio_Of_Deposit = dynamicDataConfig.Ratio_Of_Deposit;//0-100 (%) - State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; - Transaction_Length = 4; - Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; - Ratio_of_Overlapped_Keys = dynamicDataConfig.Ratio_of_Overlapped_Keys; + public void tranToDecisionConf() { + //TD,LD,PD,VDD,R_of_A,isCD,isCC, + StringBuilder stringBuilder = new StringBuilder(); + //TODO:hard code, function not sure + double td = Transaction_Length * dynamicDataConfig.getCheckpoint_interval() * (1 - (double)Ratio_Of_Deposit/100) + + ((double)Ratio_Of_Deposit/100) * 2 * dynamicDataConfig.getCheckpoint_interval(); + td = td *((double) Ratio_of_Overlapped_Keys/100); + stringBuilder.append(td); + stringBuilder.append(","); + double ld = Transaction_Length * dynamicDataConfig.getCheckpoint_interval() * (1 - (double)Ratio_Of_Deposit/100) + + ((double)Ratio_Of_Deposit/100) * 2 * dynamicDataConfig.getCheckpoint_interval(); + stringBuilder.append(ld); + stringBuilder.append(","); + double pd = (1 - (double)Ratio_Of_Deposit/100) * Transaction_Length * dynamicDataConfig.getCheckpoint_interval() * 2 * ((double) Ratio_of_Overlapped_Keys/100); + stringBuilder.append(pd); + stringBuilder.append(","); + stringBuilder.append((double) State_Access_Skewness/100); + stringBuilder.append(","); + stringBuilder.append(Ratio_of_Transaction_Aborts/10000); + stringBuilder.append(","); + if (AppConfig.isCyclic) { + stringBuilder.append("1,"); + } else { + stringBuilder.append("0,"); + } + if (AppConfig.complexity < 40000){ + stringBuilder.append("0,"); + } else { + stringBuilder.append("1,"); + } + stringBuilder.append(eventID+dynamicDataConfig.getShiftRate()*dynamicDataConfig.getCheckpoint_interval()*dynamicDataConfig.getTotalThreads()); + this.tranToDecisionConf.add(stringBuilder.toString()); + } - int nKeyState = dynamicDataConfig.getnKeyStates(); + @Override + public void switchConfiguration(String type) { + switch (type) { + case "default" : + Ratio_Of_Deposit = dynamicDataConfig.Ratio_Of_Deposit;//0-100 (%) + State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; + Transaction_Length = 4; + Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; + Ratio_of_Overlapped_Keys = dynamicDataConfig.Ratio_of_Overlapped_Keys; - // allocate levels for each key, to prevent circular. + nKeyState = dynamicDataConfig.getnKeyStates(); + // allocate levels for each key, to prevent circular. // int MAX_LEVEL = (nKeyState / dataConfig.getTotalThreads()) / 2; - int MAX_LEVEL = 256; - for (int i = 0; i < nKeyState; i++) { - idToLevel.put(i, random.nextInt(MAX_LEVEL)); + int MAX_LEVEL = 256; + for (int i = 0; i < nKeyState; i++) { + idToLevel.put(i, random.nextInt(MAX_LEVEL)); + } + // zipf state access generator + accountZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); + assetZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 123456789); + configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); + p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + break; + case "skew" : + State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; + accountZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); + assetZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 123456789); + configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); + p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); + break; + case "PD" : + Ratio_Of_Deposit = dynamicDataConfig.Ratio_Of_Deposit;//0-100 (%) + break; + case "unchanging" : + break; + default: + throw new IllegalStateException("Unexpected value: " + type); } - // zipf state access generator - accountZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 12345678); - assetZipf = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0, 123456789); - configure_store(1, (double) State_Access_Skewness / 100, dynamicDataConfig.getTotalThreads(), nKeyState); - p_generator = new FastZipfGenerator(nKeyState, (double) State_Access_Skewness / 100, 0); - + tranToDecisionConf(); } public void configure_store(double scale_factor, double theta, int tthread, int numItems) { floor_interval = (int) Math.floor(numItems / (double) tthread);//NUM_ITEMS / tthread; diff --git a/application/src/main/java/benchmark/datagenerator/apps/TP/TPTPGDynamicDataGenerator.java b/application/src/main/java/benchmark/datagenerator/apps/TP/TPTPGDynamicDataGenerator.java index 73b62df1b..a24bbc1ac 100644 --- a/application/src/main/java/benchmark/datagenerator/apps/TP/TPTPGDynamicDataGenerator.java +++ b/application/src/main/java/benchmark/datagenerator/apps/TP/TPTPGDynamicDataGenerator.java @@ -40,6 +40,7 @@ public class TPTPGDynamicDataGenerator extends DynamicWorkloadGenerator { private HashMap nGeneratedIds = new HashMap<>(); private ArrayList events; private int eventID = 0; + private boolean enableGroup; private HashMap idToLevel = new HashMap<>(); public TPTPGDynamicDataGenerator(DynamicDataGeneratorConfig dynamicDataConfig) { super(dynamicDataConfig); @@ -47,10 +48,16 @@ public TPTPGDynamicDataGenerator(DynamicDataGeneratorConfig dynamicDataConfig) { } @Override - public void switchConfiguration() { + public void tranToDecisionConf() { + + } + + @Override + public void switchConfiguration(String type) { State_Access_Skewness = dynamicDataConfig.State_Access_Skewness; Ratio_of_Transaction_Aborts = dynamicDataConfig.Ratio_of_Transaction_Aborts; Ratio_of_Overlapped_Keys = dynamicDataConfig.Ratio_of_Overlapped_Keys; + this.enableGroup=dynamicDataConfig.enableGroup; int nKeyState = dynamicDataConfig.getnKeyStates(); int MAX_LEVEL = 256; for (int i = 0; i < nKeyState; i++) { @@ -102,7 +109,13 @@ private Event randomTPEvent() { int id; if (!isUnique) { if (enable_states_partition) { - int partitionId = key_to_partition(p_generator.next()); + //Must Determine partitionId from EventId + int partitionId; + if (enableGroup) { + partitionId = eventID % dynamicDataConfig.getTotalThreads(); + } else { + partitionId = key_to_partition(p_generator.next()); + } id = getKey(partitionedKeyZipf[partitionId],partitionId,generatedKeys); } else { id = getKey(keyZipf,generatedKeys); diff --git a/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicDataGeneratorConfig.java b/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicDataGeneratorConfig.java index 6f0d955ad..58f320162 100644 --- a/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicDataGeneratorConfig.java +++ b/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicDataGeneratorConfig.java @@ -2,6 +2,7 @@ import benchmark.datagenerator.DataGeneratorConfig; import common.collections.Configuration; +import utils.AppConfig; /** * Generate dynamic workload configuration @@ -15,6 +16,8 @@ public class DynamicDataGeneratorConfig extends DataGeneratorConfig { private String app; /* Control the rate of the workload*/ private int shiftRate; + /* The workload is in which phase */ + private int phase; private int checkpoint_interval; public int State_Access_Skewness; public int NUM_ACCESS; @@ -23,6 +26,8 @@ public class DynamicDataGeneratorConfig extends DataGeneratorConfig { public int Transaction_Length; public int Ratio_Of_Deposit; public int Ratio_Of_Buying; + public boolean enableGroup; + public int Ratio_of_Multiple_State_Access_Ratio; public void initialize(Configuration config) { super.initialize(config); @@ -30,6 +35,7 @@ public void initialize(Configuration config) { this.type = config.getString("workloadType"); this.shiftRate = config.getInt("shiftRate"); this.checkpoint_interval = config.getInt("checkpoint"); + enableGroup = config.getBoolean("isGroup"); NUM_ACCESS = config.getInt("NUM_ACCESS", 0); State_Access_Skewness = config.getInt("State_Access_Skewness", 0); Ratio_of_Overlapped_Keys = config.getInt("Ratio_of_Overlapped_Keys", 0); @@ -37,6 +43,8 @@ public void initialize(Configuration config) { Transaction_Length = config.getInt("Transaction_Length", 1); Ratio_Of_Deposit = config.getInt("Ratio_Of_Deposit", 0); Ratio_Of_Buying=config.getInt("Ratio_Of_Buying",0); + Ratio_of_Multiple_State_Access_Ratio = config.getInt("Ratio_of_Multiple_State_Access_Ratio",100); + phase = 0; } public int getCheckpoint_interval() { @@ -50,9 +58,19 @@ public int getShiftRate() { * Generate the configuration based on type and application * @return */ - public boolean nextDataGeneratorConfig() { - //TODO: how to switch - return false; + public String nextDataGeneratorConfig() { + switch (type){ + case "0" : + return setConfigurationForSL(); + case "1" : + return setConfigurationForOB(); + case "2" : + return setConfigurationForGS(); + case "3" : + return setConfigurationForTP(); + default: + throw new IllegalStateException("Unexpected value: " + type); + } } public String getType() { @@ -62,4 +80,76 @@ public String getType() { public String getApp() { return app; } + + //User defined + private String setConfigurationForSL() { + switch (phase){ + case 0: + phase ++; + return "default"; + case 1: + case 4: + phase ++; + return "unchanging"; + case 2: + case 3: + phase ++; + this.State_Access_Skewness = this.State_Access_Skewness + 30; + return "skew"; + case 5: + case 6: + case 7: + phase ++; + this.Ratio_Of_Deposit = this.Ratio_Of_Deposit + 20 ; + return "skew" ; + default: + return null; + } + } + private String setConfigurationForOB() { + switch (phase){ + case 0: + phase++; + return "default"; + case 1: + phase++; + this.State_Access_Skewness = 85;// 100 + return "skew"; + case 2: + phase++; + this.Ratio_of_Transaction_Aborts = 8000;// 10000 + return "abort"; + default: + return null; + } + } + private String setConfigurationForGS() { + switch (phase){ + case 0: + phase++; + return "default"; + case 1: + phase++; + this.Transaction_Length = 8; + return "LD"; + case 2: + phase++; + this.Ratio_of_Transaction_Aborts = 8000;// 10000 + this.State_Access_Skewness = 85; + return "isCyclic"; + case 3: + phase++; + return "complexity"; + default: + return null; + } + } + private String setConfigurationForTP(){ + if (phase==0){ + phase ++; + return "default"; + } else { + return null; + } + } } diff --git a/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicWorkloadGenerator.java b/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicWorkloadGenerator.java index eb68bf7d8..279506b80 100644 --- a/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicWorkloadGenerator.java +++ b/application/src/main/java/benchmark/dynamicWorkloadGenerator/DynamicWorkloadGenerator.java @@ -1,7 +1,9 @@ package benchmark.dynamicWorkloadGenerator; import benchmark.datagenerator.DataGenerator; -import benchmark.datagenerator.DataGeneratorConfig; + +import java.util.ArrayList; +import java.util.List; /** @@ -12,33 +14,40 @@ * Created by curry on 16/3/22. */ public abstract class DynamicWorkloadGenerator extends DataGenerator { + protected List tranToDecisionConf =new ArrayList<>(); protected DynamicDataGeneratorConfig dynamicDataConfig; public DynamicWorkloadGenerator(DynamicDataGeneratorConfig dynamicDataConfig) { super(dynamicDataConfig); this.dynamicDataConfig=dynamicDataConfig; } - private String[] tranToDecisionConf() { - return null; - } + /** + * Each application may be different + */ + public abstract void tranToDecisionConf(); @Override public void generateStream() { //Init the Configuration - switchConfiguration(); for (int tupleNumber = 0; tupleNumber < nTuples + dynamicDataConfig.getTotalThreads(); tupleNumber++) { - if (tupleNumber%dynamicDataConfig.getCheckpoint_interval()* dynamicDataConfig.getShiftRate()* dynamicDataConfig.getTotalThreads() == 0) { - if (dynamicDataConfig.nextDataGeneratorConfig()) { - switchConfiguration(); + if (tupleNumber%(dynamicDataConfig.getCheckpoint_interval()* dynamicDataConfig.getShiftRate()* dynamicDataConfig.getTotalThreads()) == 0) { + String type = dynamicDataConfig.nextDataGeneratorConfig(); + if (type != null) { + switchConfiguration(type); } } generateTuple(); } } - public abstract void switchConfiguration(); + public abstract void switchConfiguration(String type); @Override public DynamicDataGeneratorConfig getDataConfig() { return dynamicDataConfig; } + + @Override + public List getTranToDecisionConf() { + return tranToDecisionConf; + } } diff --git a/application/src/main/java/common/bolts/transactional/gs/GSBolt.java b/application/src/main/java/common/bolts/transactional/gs/GSBolt.java index 3fdb0745b..b7a04ff90 100644 --- a/application/src/main/java/common/bolts/transactional/gs/GSBolt.java +++ b/application/src/main/java/common/bolts/transactional/gs/GSBolt.java @@ -37,8 +37,9 @@ protected void TXN_PROCESS(long _bid) throws DatabaseException, InterruptedExcep protected boolean READ_CORE(MicroEvent event) { for (int i = 0; i < event.TOTAL_NUM_ACCESS; ++i) { SchemaRecordRef ref = event.getRecord_refs()[i]; - if (ref.isEmpty()) + if (ref.isEmpty()) { return false;//not yet processed. + } DataBox dataBox = ref.getRecord().getValues().get(1); int read_result = Integer.parseInt(dataBox.getString().trim()); event.result[i] = read_result; diff --git a/application/src/main/java/common/bolts/transactional/gs/GSBolt_ts.java b/application/src/main/java/common/bolts/transactional/gs/GSBolt_ts.java index 66f0a7c02..d2354fd6d 100644 --- a/application/src/main/java/common/bolts/transactional/gs/GSBolt_ts.java +++ b/application/src/main/java/common/bolts/transactional/gs/GSBolt_ts.java @@ -132,9 +132,6 @@ public void execute(Tuple in) throws InterruptedException, DatabaseException, Br { MeasureTools.BEGIN_TXN_TIME_MEASURE(thread_Id); { - if (i==1){ - System.out.println(); - } transactionManager.start_evaluate(thread_Id, in.getBID(), num_events);//start lazy evaluation in transaction manager. i++; READ_REQUEST_CORE(); diff --git a/application/src/main/java/common/bolts/transactional/ob/OBBolt_ts_s.java b/application/src/main/java/common/bolts/transactional/ob/OBBolt_ts_s.java index 5b1d63b6f..e16592633 100644 --- a/application/src/main/java/common/bolts/transactional/ob/OBBolt_ts_s.java +++ b/application/src/main/java/common/bolts/transactional/ob/OBBolt_ts_s.java @@ -64,7 +64,6 @@ public void execute(Tuple in) throws InterruptedException, DatabaseException, Br BUYING_REQUEST_CORE(); } MeasureTools.END_TXN_TIME_MEASURE(thread_Id); - BUYING_REQUEST_POST(); MeasureTools.BEGIN_POST_TIME_MEASURE(thread_Id); BUYING_REQUEST_POST(); MeasureTools.END_POST_TIME_MEASURE_ACC(thread_Id); @@ -73,7 +72,7 @@ public void execute(Tuple in) throws InterruptedException, DatabaseException, Br alertEvents = 0; toppingEvents = 0; } - MeasureTools.END_TOTAL_TIME_MEASURE_TS(thread_Id,readSize); + MeasureTools.END_TOTAL_TIME_MEASURE_TS(thread_Id,num_events); }else { execute_ts_normal(in); } diff --git a/application/src/main/java/common/bolts/transactional/tp/TPBolt_ts_s.java b/application/src/main/java/common/bolts/transactional/tp/TPBolt_ts_s.java index 764a2db88..7f1b48ae2 100644 --- a/application/src/main/java/common/bolts/transactional/tp/TPBolt_ts_s.java +++ b/application/src/main/java/common/bolts/transactional/tp/TPBolt_ts_s.java @@ -21,6 +21,7 @@ import static common.CONTROL.enable_latency_measurement; import static common.constants.TPConstants.Constant.NUM_SEGMENTS; +import static profiler.Metrics.NUM_ITEMS; public class TPBolt_ts_s extends TPBolt { private static final Logger LOG= LoggerFactory.getLogger(TPBolt_ts_s.class); @@ -36,7 +37,7 @@ public TPBolt_ts_s(int fid){ public void initialize(int thread_Id, int thisTaskId, ExecutionGraph graph) { super.initialize(thread_Id, thisTaskId, graph); transactionManager=new TxnManagerTStream(db.getStorageManager(), this.context.getThisComponentId(),thread_Id, - NUM_SEGMENTS,this.context.getThisComponent().getNumTasks(),config.getString("scheduler","BF")); + NUM_ITEMS,this.context.getThisComponent().getNumTasks(),config.getString("scheduler","BF")); LREvents = new ArrayDeque<>(); } public void loadDB(Map conf, TopologyContext context, OutputCollector collector) { diff --git a/application/src/main/java/common/sink/MeasureSink.java b/application/src/main/java/common/sink/MeasureSink.java index c2ed47ba6..792ed7c93 100644 --- a/application/src/main/java/common/sink/MeasureSink.java +++ b/application/src/main/java/common/sink/MeasureSink.java @@ -170,7 +170,6 @@ protected void latency_measure(Tuple input) { // } // } latency_map.add(System.nanoTime() - input.getLong(1)); - } } diff --git a/application/src/main/java/common/topology/transactional/initializer/GSInitializer.java b/application/src/main/java/common/topology/transactional/initializer/GSInitializer.java index 880fbd6cf..7d3db4462 100644 --- a/application/src/main/java/common/topology/transactional/initializer/GSInitializer.java +++ b/application/src/main/java/common/topology/transactional/initializer/GSInitializer.java @@ -8,6 +8,7 @@ import benchmark.datagenerator.apps.GS.TPGTxnGenerator.GSTPGDynamicDataGenerator; import benchmark.datagenerator.apps.SL.TPGTxnGenerator.SLTPGDataGeneratorConfig; import benchmark.dynamicWorkloadGenerator.DynamicDataGeneratorConfig; +import benchmark.dynamicWorkloadGenerator.DynamicWorkloadGenerator; import common.collections.Configuration; import common.collections.OsUtils; import common.param.TxnEvent; @@ -300,7 +301,7 @@ private void loadMicroEvents(BufferedReader reader, int totalEvents, boolean shu String[] split = txn.split(","); int npid = (int) (Long.parseLong(split[1]) / partitionOffset); // construct bid array - int keyLength = NUM_ACCESS*Transaction_Length; + int keyLength = split.length - 2; HashMap pids = new HashMap<>(); long[] keys = new long[keyLength]; for (int i = 1; i < keyLength+1; i++) { @@ -382,6 +383,11 @@ public void store(String file_name) throws IOException { w.close(); } + @Override + public List getTranToDecisionConf() { + return dataGenerator.getTranToDecisionConf(); + } + private RecordSchema MicroTableSchema() { List dataBoxes = new ArrayList<>(); List fieldNames = new ArrayList<>(); @@ -397,6 +403,15 @@ public void creates_Table(Configuration config) { db.createTable(s, "MicroTable"); try { prepare_input_events(config.getInt("totalEvents")); + if (getTranToDecisionConf() != null && getTranToDecisionConf().size() !=0){ + StringBuilder stringBuilder = new StringBuilder(); + for(String decision:getTranToDecisionConf()){ + stringBuilder.append(decision); + stringBuilder.append(";"); + } + stringBuilder.deleteCharAt(stringBuilder.length()-1); + config.put("WorkloadConfig",stringBuilder.toString()); + } } catch (IOException e) { e.printStackTrace(); } diff --git a/application/src/main/java/common/topology/transactional/initializer/OBInitializer.java b/application/src/main/java/common/topology/transactional/initializer/OBInitializer.java index 00a0d30b8..30943bca6 100644 --- a/application/src/main/java/common/topology/transactional/initializer/OBInitializer.java +++ b/application/src/main/java/common/topology/transactional/initializer/OBInitializer.java @@ -227,7 +227,7 @@ private void loadOnlineBiddingEvents(BufferedReader reader, int totalEvents, boo String[] split = txn.split(","); if (split.length > 3) { int npid = (int) (Long.parseLong(split[1]) / partitionOffset); - int keyLength = NUM_ACCESS*Transaction_Length; + int keyLength = 1*Transaction_Length; HashMap pids = new HashMap<>(); int[] keys = new int[keyLength]; for (int i = 1; i < keyLength+1; i++) { @@ -338,11 +338,25 @@ public void store(String file_name) throws IOException { w.close(); } + @Override + public List getTranToDecisionConf() { + return dataGenerator.getTranToDecisionConf(); + } + public void creates_Table(Configuration config) { RecordSchema s = Goods(); db.createTable(s, "goods"); try { prepare_input_events(config.getInt("totalEvents")); + if (getTranToDecisionConf() != null && getTranToDecisionConf().size() != 0){//input data already exist + StringBuilder stringBuilder = new StringBuilder(); + for(String decision:getTranToDecisionConf()){ + stringBuilder.append(decision); + stringBuilder.append(";"); + } + stringBuilder.deleteCharAt(stringBuilder.length()-1); + config.put("WorkloadConfig",stringBuilder.toString()); + } } catch (IOException e) { e.printStackTrace(); } diff --git a/application/src/main/java/common/topology/transactional/initializer/SLInitializer.java b/application/src/main/java/common/topology/transactional/initializer/SLInitializer.java index 992d1aa54..e216336f5 100644 --- a/application/src/main/java/common/topology/transactional/initializer/SLInitializer.java +++ b/application/src/main/java/common/topology/transactional/initializer/SLInitializer.java @@ -325,6 +325,11 @@ public void store(String file_name) throws IOException { } + @Override + public List getTranToDecisionConf() { + return dataGenerator.getTranToDecisionConf(); + } + private void loadTransferDepositEvents(BufferedReader reader, int totalEvents, boolean shufflingActive, int[] p_bids) throws IOException { String txn = reader.readLine(); int count = 0; @@ -417,6 +422,15 @@ public void creates_Table(Configuration config) { db.createTable(b, "bookEntries"); try { prepare_input_events(config.getInt("totalEvents")); + if (getTranToDecisionConf() != null && getTranToDecisionConf().size() != 0){ + StringBuilder stringBuilder = new StringBuilder(); + for(String decision:getTranToDecisionConf()){ + stringBuilder.append(decision); + stringBuilder.append(";"); + } + stringBuilder.deleteCharAt(stringBuilder.length()-1); + config.put("WorkloadConfig",stringBuilder.toString()); + } } catch (IOException e) { e.printStackTrace(); } diff --git a/application/src/main/java/common/topology/transactional/initializer/TPInitializer.java b/application/src/main/java/common/topology/transactional/initializer/TPInitializer.java index 8d6f06072..0f1aca6a3 100644 --- a/application/src/main/java/common/topology/transactional/initializer/TPInitializer.java +++ b/application/src/main/java/common/topology/transactional/initializer/TPInitializer.java @@ -294,6 +294,11 @@ private void shuffleEvents(ArrayList txnEvents, int totalEvents) { public void store(String file_name) { } + @Override + public List getTranToDecisionConf() { + return dataGenerator.getTranToDecisionConf(); + } + protected String getConfigKey(String template) { return String.format(template, "tptxn");//TODO: make it flexible in future. } @@ -305,6 +310,15 @@ public void creates_Table(Configuration config) { db.createTable(b, "segment_cnt"); try { prepare_input_events(config.getInt("totalEvents")); + if (getTranToDecisionConf() != null && getTranToDecisionConf().size() != 0){ + StringBuilder stringBuilder = new StringBuilder(); + for(String decision:getTranToDecisionConf()){ + stringBuilder.append(decision); + stringBuilder.append(";"); + } + stringBuilder.deleteCharAt(stringBuilder.length()-1); + config.put("WorkloadConfig",stringBuilder.toString()); + } } catch (IOException e) { e.printStackTrace(); } diff --git a/application/src/main/java/runners/MorphStreamRunner.java b/application/src/main/java/runners/MorphStreamRunner.java index 93acc10cc..f31d20447 100644 --- a/application/src/main/java/runners/MorphStreamRunner.java +++ b/application/src/main/java/runners/MorphStreamRunner.java @@ -305,7 +305,7 @@ public void run() throws InterruptedException { } catch (IOException e) { e.printStackTrace(); } - METRICS_REPORT(config.getInt("CCOption", 0), file, tthread, rt); + METRICS_REPORT(config.getInt("CCOption", 0), file, tthread, rt,config.getInt("phaseNum")); } }//end of profile. } diff --git a/common/src/main/java/common/Runner.java b/common/src/main/java/common/Runner.java index 5bd3a98dc..12ddef274 100644 --- a/common/src/main/java/common/Runner.java +++ b/common/src/main/java/common/Runner.java @@ -19,10 +19,10 @@ public abstract class Runner implements IRunner { * Workload Specific Parameters. */ @Parameter(names = {"-a", "--app"}, description = "The application to be executed") - //public String application = "StreamLedger"; + public String application = "StreamLedger"; //public String application = "GrepSum"; //public String application = "OnlineBiding"; - public String application = "TollProcessing"; + //public String application = "TollProcessing"; @Parameter(names = {"-t", "--topology-name"}, required = false, description = "The name of the application") public String topologyName; @Parameter(names = {"--COMPUTE_COMPLEXITY"}, description = "COMPUTE_COMPLEXITY per event") @@ -33,7 +33,7 @@ public abstract class Runner implements IRunner { public int NUM_ITEMS = 100_000;// // public int NUM_ITEMS = 500;// @Parameter(names = {"--NUM_ACCESS"}, description = "Number of state access per transaction") - public int NUM_ACCESS = 10;// + public int NUM_ACCESS = 5;// @Parameter(names = {"--ratio_of_read"}, description = "ratio_of_read") public double ratio_of_read = 0.0; //<=1 @Parameter(names = {"--ratio_of_multi_partition"}, description = "ratio_of_multi_partition") @@ -67,8 +67,8 @@ public abstract class Runner implements IRunner { @Parameter(names = {"--partition"}, description = "Partitioning database. It must be enabled for S-Store scheme and it is optional for TStream scheme.") public boolean enable_partition = false; @Parameter(names = {"--scheduler"}, description = "Scheduler for TStream.") - public String scheduler = "OP_BFS"; - // public String scheduler = "OG_BFS_A"; + public String scheduler = "OP_BFS_A"; + //public String scheduler = "OG_BFS_A"; // public String scheduler = "OG_DFS"; // public String scheduler = "OG_DFS_A"; // public String scheduler = "OG_NS"; @@ -91,13 +91,13 @@ public abstract class Runner implements IRunner { @Parameter(names = {"--isRuntime"}, description = "Collect runtime information") public boolean isRuntime = false; @Parameter(names = {"--isDynamic"}, description = "Dynamic Workload") - public boolean isDynamic = false; + public int isDynamic = 0; @Parameter(names = {"--schedulerPool"}, description = "Schedulers in the SchedulerPool[OG_DFS,OP_DFS]") - public String schedulerPools = "OP_NS_A,OP_BFS_A,OP_BFS,OG_BFS"; + public String schedulerPools = ""; @Parameter(names = {"--defaultScheduler"}, description = "Default scheduler") - public String defaultScheduler = "OG_BFS"; + public String defaultScheduler = ""; @Parameter(names = {"--bottomLine"}, description = "BottomLine for(TD,LD,PD,SUM,VDD,R_of_A)") - public String bottomLine = "2,3,5,10,0.6,0.7"; + public String bottomLine = ""; @Parameter(names = {"--WorkloadConfig"}, description = "WorkloadConfigs(TD,LD,PD,VDD,R_of_A,isCD,isCC,markId)") public String WorkloadConfig = "1,3,4,0.5,0.6,1,1,39999;1,6,4,0.5,0.7,1,1,79999;3,6,3,0.5,0.7,1,1,100000"; @@ -105,10 +105,21 @@ public abstract class Runner implements IRunner { * Dynamic workload */ @Parameter(names = {"--workloadType"}, description = "which type of dynamic workload") - public String workloadType = ""; - @Parameter(names = {"-shiftRate-"}, description = "control the rate of the workload shift") - public int shiftRate = 5; + @Parameter(names = {"--shiftRate"}, description = "control the rate of the workload shift") + public int shiftRate = 1; + @Parameter(names = {"--phaseNum"}, description = "How many phases") + public int phaseNum = 2; + + /** + * Scheduler for group + */ + @Parameter(names = {"--isGroup"}, description = "Group for TP") + public int isGroup = 0; + @Parameter(names = {"--groupNum"}, description = "How many groups") + public int groupNum = 2; + @Parameter(names = {"--SchedulersForGroup"}, description = "Schedulers [OG_DFS,OP_DFS]") + public String SchedulersForGroup = ""; /** * Benchmarking Specific Parameters. @@ -148,11 +159,14 @@ public abstract class Runner implements IRunner { @Parameter(names = {"--key_skewness"}, description = "State access skewness.") public Integer State_Access_Skewness = 0; + @Parameter(names = {"--multiple_ratio"}, description = "State access skewness.") + public Integer Ratio_of_Multiple_State_Access_Ratio = 100; + @Parameter(names = {"--overlap_ratio"}, description = "Ratio of overlapped keys.") public Integer Ratio_of_Overlapped_Keys = 10; @Parameter(names = {"--abort_ratio"}, description = "Ratio of transaction aborts.") - public Integer Ratio_of_Transaction_Aborts = 0; + public Integer Ratio_of_Transaction_Aborts = 500; @Parameter(names = {"--txn_length"}, description = "Transaction Length.") public Integer Transaction_Length = 1; @@ -164,7 +178,7 @@ public abstract class Runner implements IRunner { public int isCyclic = 0; @Parameter(names = {"--complexity"}, description = "Dummy UDF complexity for state access process.") - public Integer complexity = 100000; + public Integer complexity = 10000; public Runner() { CFG_PATH = "/config/%s.properties"; @@ -188,7 +202,6 @@ public void initializeCfg(HashMap config) { config.put("machine", machine); config.put("totalEvents", totalEvents); config.put("rootFilePath", rootPath); - config.put("scheduler", scheduler); config.put("generator", generator); config.put("fanoutDist", fanoutDist); config.put("idGenType", idGenType); @@ -201,6 +214,7 @@ public void initializeCfg(HashMap config) { config.put("State_Access_Skewness", State_Access_Skewness); config.put("Ratio_of_Overlapped_Keys", Ratio_of_Overlapped_Keys); config.put("Ratio_of_Transaction_Aborts", Ratio_of_Transaction_Aborts); + config.put("Ratio_of_Multiple_State_Access_Ratio",Ratio_of_Multiple_State_Access_Ratio); config.put("Transaction_Length", Transaction_Length); config.put("numberOfDLevels", numberOfDLevels); @@ -240,18 +254,66 @@ public void initializeCfg(HashMap config) { config.put("size_tuple", size_tuple); config.put("verbose", verbose); + config.put("application",application); + switch(application) { + case "StreamLedger" : + workloadType = "0"; + bottomLine = "300,3000,500,1200,0.2,0.2";//TD,LD,PD,SUM,VDD,R_of_A + schedulerPools = "OP_BFS_A,OP_NS_A,OG_NS_A"; + defaultScheduler = "OP_BFS_A"; + phaseNum = shiftRate * 8; + break; + case "OnlineBiding" : + workloadType ="1"; + bottomLine = "500,5000,1,6000,0.2,0.2";//TD,LD,PD,SUM,VDD,R_of_A + schedulerPools = "OG_BFS_A,OG_NS_A,OG_NS"; + defaultScheduler = "OG_BFS_A"; + phaseNum = shiftRate * 3; + break; + case "GrepSum" : + workloadType ="2"; + bottomLine = "500,5000,6500,3000,0.2,0.2";//TD,LD,PD,SUM,VDD,R_of_A + schedulerPools = "OP_NS_A,OG_BFS_A,OP_NS,OP_NS_A"; + defaultScheduler = "OP_NS_A"; + phaseNum = shiftRate * 4; + break; + case "TollProcessing" : + workloadType ="3"; + phaseNum = shiftRate * 1; + defaultScheduler = "OG_BFS_A"; + SchedulersForGroup = "OP_BFS_A,OP_NS_A"; + break; + default: + throw new IllegalStateException("Unexpected value: " + application); + } /* Dynamic switch scheduler*/ - config.put("isDynamic",isDynamic); + if (isDynamic == 1) { + config.put("isDynamic", true); + config.put("totalEvents",phaseNum * tthread * checkpoint_interval); + } else { + config.put("isDynamic", false); + } config.put("schedulersPool",schedulerPools); - config.put("defaultScheduler",scheduler); + config.put("defaultScheduler",defaultScheduler); + config.put("scheduler", defaultScheduler); config.put("isRuntime",isRuntime); config.put("bottomLine",bottomLine); config.put("WorkloadConfig",WorkloadConfig); /* Dynamic Workload Configuration*/ - config.put("application",application); config.put("workloadType",workloadType); config.put("shiftRate",shiftRate); + config.put("phaseNum",phaseNum); + + /* Group scheduler*/ + if (isGroup == 1) { + config.put("isGroup", true); + config.put("groupNum",groupNum); + config.put("SchedulersForGroup",SchedulersForGroup); + config.put("totalEvents",phaseNum * tthread * checkpoint_interval); + } else { + config.put("isGroup", false); + } System.setProperty("my.log", metric_path); } diff --git a/common/src/main/java/common/platform/Platform.java b/common/src/main/java/common/platform/Platform.java index 56e2b0a09..1146f2e93 100644 --- a/common/src/main/java/common/platform/Platform.java +++ b/common/src/main/java/common/platform/Platform.java @@ -92,7 +92,7 @@ public static ArrayList[] getNodes(int machine) { node_5, node_6, node_7}; - } else { + } else if (machine == 3){ // //a simple single-socket 48-core machine. // int cores = 48; // Integer[] no_0 = new Integer[cores]; @@ -110,6 +110,16 @@ public static ArrayList[] getNodes(int machine) { return new ArrayList[]{ node_0, node_1}; + } else { + //a two-sockets 40-core machine + //Intel(R) Xeon(R) Gold 6230 CPU @ 2.10GHz + Integer[] no_0 = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19}; + node_0 = new ArrayList<>(Arrays.asList(no_0)); + Integer[] no_1 = {20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39}; + ArrayList node_1 = new ArrayList<>(Arrays.asList(no_1)); + return new ArrayList[]{ + node_0, + node_1}; } } } \ No newline at end of file diff --git a/state-engine/src/main/java/profiler/MeasureTools.java b/state-engine/src/main/java/profiler/MeasureTools.java index 012d8961f..c25417b96 100644 --- a/state-engine/src/main/java/profiler/MeasureTools.java +++ b/state-engine/src/main/java/profiler/MeasureTools.java @@ -8,6 +8,8 @@ import java.io.File; import java.nio.file.Files; import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import static common.CONTROL.enable_debug; @@ -362,10 +364,34 @@ private static void WriteThroughputReport(File file, double throughput) { e.printStackTrace(); } } + private static void WriteThroughputReportRuntime(File file, int tthread, int phase) { + try { + double[] tr = new double[Metrics.Runtime.ThroughputPerPhase.get(0).size()]; + for (int i = 0; i < tr.length; i++) { + tr[i] = 0; + for (int j = 0; j < tthread; j++){ + tr[i] = tr[i] + Metrics.Runtime.ThroughputPerPhase.get(j).get(i) * 1E6;// + } + } + BufferedWriter fileWriter = Files.newBufferedWriter(Paths.get(file.getPath()), APPEND); + fileWriter.write("phase_id\t throughput\n"); + for (int i = 0; i < tr.length; i++){ + String output = String.format("%d\t" + + "%-10.4f\t" + , i,tr[i] + ); + fileWriter.write(output + "\n"); + } + fileWriter.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } - public static void METRICS_REPORT(int ccOption, File file, int tthread, double throughput) { + public static void METRICS_REPORT(int ccOption, File file, int tthread, double throughput, int phase) { WriteThroughputReport(file, throughput); AverageTotalTimeBreakdownReport(file, tthread); + WriteThroughputReportRuntime(file,tthread,phase); if (ccOption == CCOption_MorphStream) {//extra info SchedulerTimeBreakdownReport(file, tthread); } else { diff --git a/state-engine/src/main/java/profiler/Metrics.java b/state-engine/src/main/java/profiler/Metrics.java index 46176c087..c92cfdc12 100644 --- a/state-engine/src/main/java/profiler/Metrics.java +++ b/state-engine/src/main/java/profiler/Metrics.java @@ -2,6 +2,11 @@ import org.apache.commons.math.stat.descriptive.DescriptiveStatistics; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + import static content.common.CommonMetaTypes.kMaxThreadNum; public class Metrics { @@ -55,6 +60,8 @@ public static void RECORD_TIME(int thread_id, int number_events) { Total_Record.stream_total[thread_id].addValue(stream_total); Total_Record.txn_total[thread_id].addValue(txn_total); Total_Record.overhead_total[thread_id].addValue(total_time - stream_total - txn_total); + Runtime.ThroughputPerPhase.get(thread_id).add(1 / total_time); + Runtime.Start[thread_id]=0; } public static void COMPUTE_PRE_EXE_START_TIME(int thread_id) { @@ -275,6 +282,8 @@ static class Runtime { public static long[] Post = new long[kMaxThreadNum]; public static long[] TxnStart = new long[kMaxThreadNum]; public static long[] Txn = new long[kMaxThreadNum]; + //Runtime throughput per phase + public static HashMap> ThroughputPerPhase =new HashMap<>(); //USED ONLY BY TStream public static long[] PreTxnStart = new long[kMaxThreadNum]; public static long[] PreTxn = new long[kMaxThreadNum]; @@ -290,6 +299,7 @@ public static void Initialize() { Txn[i] = 0; PreTxnStart[i] = 0; PreTxn[i] = 0; + ThroughputPerPhase.put(i,new ArrayList<>()); } } } diff --git a/state-engine/src/main/java/scheduler/collector/Collector.java b/state-engine/src/main/java/scheduler/collector/Collector.java index 6c9a400c7..d1927562b 100644 --- a/state-engine/src/main/java/scheduler/collector/Collector.java +++ b/state-engine/src/main/java/scheduler/collector/Collector.java @@ -12,10 +12,10 @@ public class Collector { Num of TD,LD,PD */ private HashMap Num_of_dependency; - private int B_TD; - private int B_LD; - private int B_PD; - private int B_SUM_D; + private double B_TD; + private double B_LD; + private double B_PD; + private double B_SUM_D; /* Vertex Degree Distribution */ @@ -37,7 +37,7 @@ public class Collector { private boolean isRuntime = true; private int threadCount; private final List workloadConfig=new ArrayList<>(); - private final HashMap currentWorkload=new HashMap<>(); + private final HashMap currentWorkload=new HashMap<>();// /** * Init the collector */ @@ -52,12 +52,12 @@ public void InitCollector(int threadCount){ */ public void setBottomLine(String bottomLine){ String b_ls[]=bottomLine.split(","); - this.B_TD= Integer.parseInt(b_ls[0]); - this.B_LD= Integer.parseInt(b_ls[1]); - this.B_PD= Integer.parseInt(b_ls[2]); - this.B_SUM_D=Integer.parseInt(b_ls[3]); - this.B_VDD= Double.parseDouble(b_ls[4]); - this.B_R_of_A= Double.parseDouble(b_ls[5]); + this.B_TD = Double.parseDouble(b_ls[0]); + this.B_LD = Double.parseDouble(b_ls[1]); + this.B_PD = Double.parseDouble(b_ls[2]); + this.B_SUM_D = Double.parseDouble(b_ls[3]); + this.B_VDD = Double.parseDouble(b_ls[4]); + this.B_R_of_A = Double.parseDouble(b_ls[5]); } /** * Load the workload config if not collecting information runtime @@ -79,7 +79,7 @@ public void setWorkloadConfig(String config){ * 0000 0001 0010 0011 0100 0101 0110 0111 1000 1001 1010 1011 * @return */ - private String decisionTree(int TD,int LD,int PD,double VDD,boolean isCyclicDependency,boolean isComputationComplexity,double R_of_A){ + private String decisionTree(double TD,double LD,double PD,double VDD,boolean isCyclicDependency,boolean isComputationComplexity,double R_of_A){ String schedulers[]={"OG_NS","OG_NS_A","OP_NS","OP_NS_A","OG_BFS","OG_BFS_A","OP_BFS","OP_BFS_A","OG_DFS","OG_DFS_A","OP_DFS","OP_DFS_A"}; int flag=0; if((TD+PD+LD)>B_SUM_D){ @@ -100,11 +100,11 @@ private String decisionTree(int TD,int LD,int PD,double VDD,boolean isCyclicDepe private String getDecisionFromConf(int threadId){ int workloadId=currentWorkload.get(threadId); String para[]=workloadConfig.get(workloadId); - int TD= Integer.parseInt(para[0]); - int LD= Integer.parseInt(para[1]); - int PD= Integer.parseInt(para[2]); - double VDD=Double.parseDouble(para[3]); - double R_of_A=Double.parseDouble(para[4]); + double TD = Double.parseDouble(para[0]); + double LD = Double.parseDouble(para[1]); + double PD = Double.parseDouble(para[2]); + double VDD = Double.parseDouble(para[3]); + double R_of_A = Double.parseDouble(para[4]); boolean isCD = false,isCC=false; if(para[5].equals("1")){ isCD=true; @@ -126,18 +126,24 @@ public String getDecision(int threadId){ } } - public boolean timeToSwitch(long markId,int threadId){ + public boolean timeToSwitch(long markId,int threadId,String currentScheduler){ if(isRuntime){ //TODO:collect information runtime return false; }else{ - int workloadId=this.currentWorkload.get(threadId); - long currentId= Long.parseLong(workloadConfig.get(workloadId)[workloadConfig.get(workloadId).length-1]); - if (currentId<=markId+threadCount-threadId-1){ - currentWorkload.put(threadId,currentWorkload.get(threadId)+1); - return true; - }else - return false; + int workloadId = this.currentWorkload.get(threadId); + if (workloadId < workloadConfig.size() -1){ + long currentId = Long.parseLong(workloadConfig.get(workloadId)[workloadConfig.get(workloadId).length-1]); + if (currentId <= markId+threadCount-threadId){ + currentWorkload.put(threadId,currentWorkload.get(threadId)+1); + if(getDecision(threadId).equals(currentScheduler)){ + return false; + } else { + return true; + } + } + } + return false; } } } diff --git a/state-engine/src/main/java/scheduler/impl/IScheduler.java b/state-engine/src/main/java/scheduler/impl/IScheduler.java index ed8063e0f..26125ce3f 100644 --- a/state-engine/src/main/java/scheduler/impl/IScheduler.java +++ b/state-engine/src/main/java/scheduler/impl/IScheduler.java @@ -23,4 +23,5 @@ public interface IScheduler { void start_evaluation(Context context, long mark_ID, int num_events); + void initTPG(int offset); } diff --git a/state-engine/src/main/java/scheduler/impl/og/OGScheduler.java b/state-engine/src/main/java/scheduler/impl/og/OGScheduler.java index 6df68db0d..10816ad3b 100644 --- a/state-engine/src/main/java/scheduler/impl/og/OGScheduler.java +++ b/state-engine/src/main/java/scheduler/impl/og/OGScheduler.java @@ -41,6 +41,11 @@ protected OGScheduler(int totalThreads, int NUM_ITEMS, int app) { this.tpg = new TaskPrecedenceGraph<>(totalThreads, delta, NUM_ITEMS, app); } + @Override + public void initTPG(int offset) { + tpg.initTPG(offset); + } + /** * state to thread mapping * @@ -450,4 +455,5 @@ protected boolean isConflicted(Context context, OperationChain operationChain, O } return false; } + } diff --git a/state-engine/src/main/java/scheduler/impl/og/structured/OGSScheduler.java b/state-engine/src/main/java/scheduler/impl/og/structured/OGSScheduler.java index a5f2ecec7..fba911c48 100644 --- a/state-engine/src/main/java/scheduler/impl/og/structured/OGSScheduler.java +++ b/state-engine/src/main/java/scheduler/impl/og/structured/OGSScheduler.java @@ -20,7 +20,7 @@ public void INITIALIZE(Context context) { int threadId = context.thisThreadId; // tpg.constructTPG(context); tpg.firstTimeExploreTPG(context); - SOURCE_CONTROL.getInstance().preStateAccessBarrier(threadId);//sync for all threads to come to this line to ensure chains are constructed for the current batch. + SOURCE_CONTROL.getInstance().exploreTPGBarrier(threadId);//sync for all threads to come to this line to ensure chains are constructed for the current batch. } protected void ProcessedToNextLevel(Context context) { diff --git a/state-engine/src/main/java/scheduler/impl/op/OPScheduler.java b/state-engine/src/main/java/scheduler/impl/op/OPScheduler.java index aa57b55a1..17b4e371a 100644 --- a/state-engine/src/main/java/scheduler/impl/op/OPScheduler.java +++ b/state-engine/src/main/java/scheduler/impl/op/OPScheduler.java @@ -32,12 +32,15 @@ public abstract class OPScheduler impl private static final Logger log = LoggerFactory.getLogger(OPScheduler.class); public final int delta;//range of each partition. depends on the number of op in the stage. public final TaskPrecedenceGraph tpg; // TPG to be maintained in this global instance. - + public int groupNum; public OPScheduler(int totalThreads, int NUM_ITEMS, int app) { delta = (int) Math.ceil(NUM_ITEMS / (double) totalThreads); // Check id generation in DateGenerator. this.tpg = new TaskPrecedenceGraph<>(totalThreads, delta, NUM_ITEMS, app); } - + @Override + public void initTPG(int offset) { + tpg.initTPG(offset); + } /** * state to thread mapping * @@ -339,4 +342,5 @@ public void start_evaluation(Context context, long mark_ID, int num_events) { RESET(context);// // MeasureTools.SCHEDULE_TIME_RECORD(threadId, num_events); } + } diff --git a/state-engine/src/main/java/scheduler/struct/og/TableOCs.java b/state-engine/src/main/java/scheduler/struct/og/TableOCs.java index 972cdcf6d..acaf79856 100644 --- a/state-engine/src/main/java/scheduler/struct/og/TableOCs.java +++ b/state-engine/src/main/java/scheduler/struct/og/TableOCs.java @@ -8,10 +8,10 @@ public class TableOCs { public ConcurrentHashMap> threadOCsMap = new ConcurrentHashMap<>();//each op has a holder. - public TableOCs(Integer num_op) { + public TableOCs(Integer num_op,int offset) { int i; for (i = 0; i < num_op; i++) { - threadOCsMap.put(i, new Holder()); + threadOCsMap.put(i + offset, new Holder()); } } } diff --git a/state-engine/src/main/java/scheduler/struct/og/TaskPrecedenceGraph.java b/state-engine/src/main/java/scheduler/struct/og/TaskPrecedenceGraph.java index 2b7155b2a..9daac1cd1 100644 --- a/state-engine/src/main/java/scheduler/struct/og/TaskPrecedenceGraph.java +++ b/state-engine/src/main/java/scheduler/struct/og/TaskPrecedenceGraph.java @@ -9,6 +9,7 @@ import scheduler.context.og.OGSContext; import scheduler.context.og.OGSchedulerContext; import scheduler.struct.op.MetaTypes; +import transaction.TxnManager; import transaction.impl.ordered.MyList; import utils.AppConfig; import utils.SOURCE_CONTROL; @@ -86,16 +87,18 @@ public TaskPrecedenceGraph(int totalThreads, int delta, int NUM_ITEMS, int app) this.app = app; //create holder. operationChains = new ConcurrentHashMap<>(); + } + public void initTPG(int offset) { if (app == 0) {//GS - operationChains.put("MicroTable", new TableOCs<>(totalThreads)); + operationChains.put("MicroTable", new TableOCs<>(totalThreads,offset)); } else if (app == 1) {//SL - operationChains.put("accounts", new TableOCs<>(totalThreads)); - operationChains.put("bookEntries", new TableOCs<>(totalThreads)); + operationChains.put("accounts", new TableOCs<>(totalThreads,offset)); + operationChains.put("bookEntries", new TableOCs<>(totalThreads,offset)); } else if(app == 2) {//TP - operationChains.put("segment_speed",new TableOCs<>(totalThreads)); - operationChains.put("segment_cnt",new TableOCs<>(totalThreads)); + operationChains.put("segment_speed",new TableOCs<>(totalThreads,offset)); + operationChains.put("segment_cnt",new TableOCs<>(totalThreads,offset)); } else if (app == 3) { - operationChains.put("goods",new TableOCs<>(totalThreads)); + operationChains.put("goods",new TableOCs<>(totalThreads,offset)); } } diff --git a/state-engine/src/main/java/scheduler/struct/op/TableOCs.java b/state-engine/src/main/java/scheduler/struct/op/TableOCs.java index da13dfec2..f64900388 100644 --- a/state-engine/src/main/java/scheduler/struct/op/TableOCs.java +++ b/state-engine/src/main/java/scheduler/struct/op/TableOCs.java @@ -8,10 +8,10 @@ public class TableOCs { public ConcurrentHashMap threadOCsMap = new ConcurrentHashMap<>();//each op has a holder. - public TableOCs(Integer num_op) { + public TableOCs(Integer num_op,int offset) { int i; for (i = 0; i < num_op; i++) { - threadOCsMap.put(i, new Holder()); + threadOCsMap.put(i+offset, new Holder()); } } } diff --git a/state-engine/src/main/java/scheduler/struct/op/TaskPrecedenceGraph.java b/state-engine/src/main/java/scheduler/struct/op/TaskPrecedenceGraph.java index c0afbde3b..3f19ae559 100644 --- a/state-engine/src/main/java/scheduler/struct/op/TaskPrecedenceGraph.java +++ b/state-engine/src/main/java/scheduler/struct/op/TaskPrecedenceGraph.java @@ -43,17 +43,17 @@ public class TaskPrecedenceGraph { public final ConcurrentHashMap> threadToOCs; public void reset(Context context) { - if (app == 0) { - operationChains.get("MicroTable").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); - } else if (app == 1) { - operationChains.get("accounts").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); - operationChains.get("bookEntries").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); - } else if (app == 2) { - operationChains.get("segment_speed").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); - operationChains.get("segment_cnt").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); - } else if (app == 3) { - operationChains.get("goods").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); - } +// if (app == 0) { +// operationChains.get("MicroTable").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); +// } else if (app == 1) { +// operationChains.get("accounts").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); +// operationChains.get("bookEntries").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); +// } else if (app == 2) { +// operationChains.get("segment_speed").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); +// operationChains.get("segment_cnt").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); +// } else if (app == 3) { +// operationChains.get("goods").threadOCsMap.get(context.thisThreadId).holder_v1.clear(); +// } // threadToOCs.get(context.thisThreadId).clear(); // for (OperationChain oc : threadToOCs.get(context.thisThreadId)) { // oc.clear(); @@ -78,16 +78,18 @@ public TaskPrecedenceGraph(int totalThreads, int delta, int NUM_ITEMS, int app) this.app = app; //create holder. operationChains = new ConcurrentHashMap<>(); + } + public void initTPG(int offset) { if (app == 0) {//GS - operationChains.put("MicroTable", new TableOCs(totalThreads)); + operationChains.put("MicroTable", new TableOCs(totalThreads,offset)); } else if (app == 1) {//SL - operationChains.put("accounts", new TableOCs(totalThreads)); - operationChains.put("bookEntries", new TableOCs(totalThreads)); + operationChains.put("accounts", new TableOCs(totalThreads,offset)); + operationChains.put("bookEntries", new TableOCs(totalThreads,offset)); } else if (app == 2){//TP - operationChains.put("segment_speed",new TableOCs(totalThreads)); - operationChains.put("segment_cnt",new TableOCs(totalThreads)); + operationChains.put("segment_speed",new TableOCs(totalThreads,offset)); + operationChains.put("segment_cnt",new TableOCs(totalThreads,offset)); } else if (app == 3) {//OB - operationChains.put("goods",new TableOCs(totalThreads)); + operationChains.put("goods",new TableOCs(totalThreads,offset)); } } @@ -126,7 +128,7 @@ public void setOCs(Context context) { operationChains.get("segment_cnt").threadOCsMap.get(context.thisThreadId).holder_v1.put(_key, cntOC); ocs.add(speedOC); ocs.add(cntOC); - } else if (app ==3) { + } else if (app == 3) { OperationChain gsOC = context.createTask("goods", _key); operationChains.get("goods").threadOCsMap.get(context.thisThreadId).holder_v1.put(_key, gsOC); ocs.add(gsOC); diff --git a/state-engine/src/main/java/transaction/TableInitilizer.java b/state-engine/src/main/java/transaction/TableInitilizer.java index f9a8147f9..0dfb52c71 100644 --- a/state-engine/src/main/java/transaction/TableInitilizer.java +++ b/state-engine/src/main/java/transaction/TableInitilizer.java @@ -11,6 +11,7 @@ import java.io.IOException; import java.util.Arrays; +import java.util.List; import java.util.Set; import java.util.SplittableRandom; @@ -151,5 +152,6 @@ public void prepare_input_events(int total_events) throws IOException { protected abstract void Load() throws IOException; public abstract void store(String file_path) throws IOException; + public abstract List getTranToDecisionConf(); } diff --git a/state-engine/src/main/java/transaction/TxnManager.java b/state-engine/src/main/java/transaction/TxnManager.java index 14bd4bb6e..16e0040c2 100644 --- a/state-engine/src/main/java/transaction/TxnManager.java +++ b/state-engine/src/main/java/transaction/TxnManager.java @@ -1,5 +1,6 @@ package transaction; +import org.apache.zookeeper.txn.Txn; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scheduler.collector.Collector; @@ -29,12 +30,35 @@ public abstract class TxnManager implements ITxnManager { private static final Logger log = LoggerFactory.getLogger(TxnManager.class); protected static IScheduler scheduler; + /** + * For dynamic workload + */ protected static HashMap schedulerPool; protected static boolean enableDynamic=false; - protected static Collector collector=new Collector(); protected static String currentSchedulerType; + /** + * Scheduler for multiple workload + */ + protected static HashMap schedulerByGroup; + protected static HashMap schedulerTypeByGroup; + public static boolean enableGroup = false; + public static int groupNum; + + public static void CreateSchedulerByGroup(String schedulerType, int threadCount,int numberOfStates, int app){ + schedulerByGroup = new HashMap<>(); + schedulerTypeByGroup = new HashMap<>(); + String[] scheduler = schedulerType.split(","); + for(int i=0;i iteration; @@ -28,13 +30,14 @@ public static SOURCE_CONTROL getInstance() { return ourInstance; } - public void config(int number_threads) { + public void config(int number_threads, int groupNum) { totalThreads = number_threads; startBarrier = new CyclicBarrier(number_threads); endBarrier = new CyclicBarrier(number_threads); finalEndBarrier = new CyclicBarrier(number_threads); - switchSchedulerBarrier=new CyclicBarrier(number_threads); - dLevelEndBarrier = new Phaser(number_threads); + switchSchedulerBarrier = new CyclicBarrier(number_threads); + exploreTPGBarrier = new CyclicBarrier(number_threads / groupNum); + dLevelEndBarrier = new Phaser(number_threads / groupNum); iteration = new HashMap<>(); for (int i = 0; i < number_threads; i++) { iteration.put(i, 0); @@ -54,6 +57,13 @@ public void preStateAccessBarrier(int threadId) { } } + public void exploreTPGBarrier(int threadId) { + try { + exploreTPGBarrier.await(); + } catch (Exception e) { + e.printStackTrace(); + } + } public void postStateAccessBarrier(int threadId) { try { endBarrier.await(); diff --git a/stream-engine/src/main/java/components/operators/api/TransactionalBolt.java b/stream-engine/src/main/java/components/operators/api/TransactionalBolt.java index c1697d3ab..59d71fbac 100644 --- a/stream-engine/src/main/java/components/operators/api/TransactionalBolt.java +++ b/stream-engine/src/main/java/components/operators/api/TransactionalBolt.java @@ -173,6 +173,10 @@ public void initialize(int thread_Id, int thisTaskId, ExecutionGraph graph) { COMPUTE_COMPLEXITY = Metrics.COMPUTE_COMPLEXITY; POST_COMPUTE_COMPLEXITY = Metrics.POST_COMPUTE_COMPLEXITY; //LOG.DEBUG("NUM_ACCESSES: " + NUM_ACCESSES + " theta:" + theta); - SOURCE_CONTROL.getInstance().config(tthread); + if (config.getBoolean("isGroup")) { + SOURCE_CONTROL.getInstance().config(tthread,config.getInt("groupNum")); + } else { + SOURCE_CONTROL.getInstance().config(tthread,1); + } } } \ No newline at end of file diff --git a/stream-engine/src/main/java/controller/affinity/SequentialBinding.java b/stream-engine/src/main/java/controller/affinity/SequentialBinding.java index a6e6d5016..f99dd106a 100644 --- a/stream-engine/src/main/java/controller/affinity/SequentialBinding.java +++ b/stream-engine/src/main/java/controller/affinity/SequentialBinding.java @@ -21,7 +21,7 @@ public static int next_cpu() { ArrayList list = mapping_node[socket]; Integer core = list.get(cpu); cpu++; - if (cpu == 24) {//assume one socket 24 cores. + if (cpu == 19) {//assume one socket 24 cores. socket++; cpu = 0; } diff --git a/stream-engine/src/main/java/execution/ExecutionManager.java b/stream-engine/src/main/java/execution/ExecutionManager.java index 728f5f1cd..a94cb586a 100644 --- a/stream-engine/src/main/java/execution/ExecutionManager.java +++ b/stream-engine/src/main/java/execution/ExecutionManager.java @@ -74,6 +74,8 @@ public void distributeTasks(Configuration conf, if (!conf.getBoolean("isRuntime")) { TxnManager.setWorkloadConfig(conf.getString("WorkloadConfig")); } + } else if (conf.getBoolean("isGroup")) { + TxnManager.CreateSchedulerByGroup(conf.getString("SchedulersForGroup"),totalThread,numberOfStates,app); } else { TxnManager.CreateScheduler(schedulerType, totalThread, numberOfStates, app); }