bdware/cm
2
0
Fork 0
mirror of https://gitee.com/BDWare/cm synced 2025-04-28 15:02:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: bdware:master
Branches Tags
bdware:master
bdware:feat/consistency-sdk
bdware:v1.6.6
..
compare: bdware:v1.6.6
Branches Tags
bdware:master
bdware:feat/consistency-sdk
bdware:v1.6.6

No commits in common. "master" and "v1.6.6" have entirely different histories.
master ... v1.6.6

49 changed files with 1822 additions and 1705 deletions
Show Stats Expand all files Collapse all files

13
build.gradle
View File

@ -2,10 +2,7 @@ plugins {
id 'java' id 'java'
id 'java-library' id 'java-library'
} }
apply from: '../spotless.gradle'
repositories {
mavenCentral()
}
sourceSets { sourceSets {
main { main {
java { java {
@ -27,10 +24,14 @@ sourceSets {
sourceCompatibility = 1.8 sourceCompatibility = 1.8
repositories {
mavenCentral()
}
dependencies { dependencies {
api project(":common") api project(":common")
api 'io.prometheus:simpleclient:0.12.0' api 'io.prometheus:simpleclient:0.12.0'
api 'org.hyperic.sigar:sigar:1.6.4' api 'org.knowhowlab.osgi:sigar:1.6.5_01'
api 'com.github.ben-manes.caffeine:caffeine:2.8.8' api fileTree(dir: 'libs', include: '*.jar')
testImplementation 'junit:junit:4.13.2' testImplementation 'junit:junit:4.13.2'
} }

BIN
libs/DOIPSDK-1.0.2-SNAPSHOT.jar Normal file
View File

Binary file not shown.

28
src/main/java/org/bdware/sc/ChainOpener.java
View File

@ -1,22 +1,32 @@
package org.bdware.sc; package org.bdware.sc;
import com.google.gson.JsonElement;
import org.bdware.sc.bean.ContractRequest; import org.bdware.sc.bean.ContractRequest;
import org.bdware.sc.conn.OnHashCallback; import org.bdware.sc.conn.OnHashCallback;
public interface ChainOpener { public interface ChainOpener {
void reRegister(String doid); void reRegister(String doid);
String register(String arg); String register(String arg);
void writeContractResultToLocalAndLedger(String result, ContractClient client, void writeContractResultToLocalAndLedger(
ContractRequest contractRequest, OnHashCallback cb, long start, long l); String result,
ContractClient client,
ContractRequest contractRequest,
OnHashCallback cb, long start, long l);
void writeToChain(OnHashCallback cb, String from, String to, String data, String requestID, void writeToChain(
OnHashCallback cb,
String from,
String to,
String data,
String requestID,
String namedLedger); String namedLedger);
void writeToChainWithContract(OnHashCallback cb, String from, String to, String data, void writeToChainWithContract(
String requestID, String contractID, String namedLedger); OnHashCallback cb,
String from,
JsonElement getLedgerParams(); String to,
String data,
String requestID,
String contractID,
String namedLedger);
} }

360
src/main/java/org/bdware/sc/ContractClient.java
View File

@ -18,13 +18,11 @@ import org.bdware.sc.encrypt.HardwareInfo.OSType;
import org.bdware.sc.event.REvent.REventSemantics; import org.bdware.sc.event.REvent.REventSemantics;
import org.bdware.sc.node.AnnotationNode; import org.bdware.sc.node.AnnotationNode;
import org.bdware.sc.node.YjsType; import org.bdware.sc.node.YjsType;
import org.bdware.sc.units.MultiContractMeta;
import org.bdware.sc.util.JsonUtil; import org.bdware.sc.util.JsonUtil;
import java.io.ByteArrayOutputStream;
import java.io.File; import java.io.File;
import java.io.InputStream;
import java.io.PrintStream; import java.io.PrintStream;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field; import java.lang.reflect.Field;
import java.util.*; import java.util.*;
@ -32,6 +30,7 @@ public class ContractClient {
private static final Logger LOGGER = LogManager.getLogger(ContractClient.class); private static final Logger LOGGER = LogManager.getLogger(ContractClient.class);
public static String cmi = ""; public static String cmi = "";
public ContractMeta contractMeta; public ContractMeta contractMeta;
public boolean isDebug;
transient SocketGet get; transient SocketGet get;
int port; int port;
String pid; String pid;
@ -42,17 +41,17 @@ public class ContractClient {
long times = 0L; long times = 0L;
long traffic = 0L; long traffic = 0L;
long memory; long memory;
// boolean withInsnLimit; // 是否需要计算gas // boolean withInsnLimit; // 是否需要计算gas
ContractStatus contractStatus; ContractStatus contractStatus;
boolean isRunning; boolean isRunning;
// public boolean changeDumpPeriod(String period) { // public boolean changeDumpPeriod(String period) {
// System.out.println("[ContractClient] changeDumpPeriod " + period); // System.out.println("[ContractClient] changeDumpPeriod " + period);
// String res = get.syncGet("", "changeDumpPeriod",period); // String res = get.syncGet("", "changeDumpPeriod",period);
// if(res.equals("success")) // if(res.equals("success"))
// return true; // return true;
// return false; // return false;
// } // }
long lastUpdate = 0; long lastUpdate = 0;
ContractClient(Contract c) { ContractClient(Contract c) {
@ -68,15 +67,11 @@ public class ContractClient {
isRunning = false; isRunning = false;
} }
public ContractClient(int startPort) {
this("127.0.0.1", startPort);
}
// Once the reconnect is called, all cp in the same node will reconnect to // Once the reconnect is called, all cp in the same node will reconnect to
// current ContractManager // current ContractManager
// We only consider the. // We only consider the.
public ContractClient(String host, int startPort) { public ContractClient(int startPort) {
// LOGGER.info("ContractClient 构造器 : "); // LOGGER.info("ContractClient 构造器 : ");
contractMeta = new ContractMeta(); contractMeta = new ContractMeta();
outputTracer = new ContractPrinter(); outputTracer = new ContractPrinter();
errorTracer = new ContractPrinter(); errorTracer = new ContractPrinter();
@ -87,7 +82,7 @@ public class ContractClient {
port = startPort; port = startPort;
// LOGGER.info("ContractClient----构造器----- 端口 " + startPort); // LOGGER.info("ContractClient----构造器----- 端口 " + startPort);
get = new SocketGet(host, startPort); get = new SocketGet("127.0.0.1", startPort);
// LOGGER.info("ContractClient----构造器----- position---2"); // LOGGER.info("ContractClient----构造器----- position---2");
String cpCMI = get.syncGet("", "isContractProcess", ""); // CMI String cpCMI = get.syncGet("", "isContractProcess", ""); // CMI
@ -133,21 +128,26 @@ public class ContractClient {
} }
private void initProps() { private void initProps() {
LOGGER.info("initProps ---- position-----1"); // LOGGER.info("initProps ---- position-----1");
contractMeta.name = get.syncGet("", "getContractName", ""); contractMeta.name = get.syncGet("", "getContractName", "");
LOGGER.info("initProps ---- position-----2"); // LOGGER.info("initProps ---- position-----2");
String strEvent = get.syncGet("", "getDeclaredEvents", ""); String strEvent = get.syncGet("", "getDeclaredEvents", "");
LOGGER.debug("event: " + strEvent); LOGGER.debug("event: " + strEvent);
contractMeta.declaredEvents = JsonUtil.fromJson(strEvent, contractMeta.declaredEvents =
new TypeToken<Map<String, REventSemantics>>() {}.getType()); JsonUtil.fromJson(
LOGGER.info("initProps ---- position-----3"); strEvent, new TypeToken<Map<String, REventSemantics>>() {
contractMeta.dependentContracts = }.getType());
JsonUtil.fromJson(get.syncGet("", "getDependentContracts", ""), // LOGGER.info("initProps ---- position-----3");
new TypeToken<Set<String>>() {}.getType()); contractMeta.dependentContracts = JsonUtil.fromJson(
LOGGER.info("initProps ---- position-----4"); get.syncGet("", "getDependentContracts", ""),
new TypeToken<Set<String>>() {
}.getType());
// LOGGER.info("initProps ---- position-----4");
contractMeta.exportedFunctions = contractMeta.exportedFunctions =
JsonUtil.fromJson(get.syncGet("", "getExportedFunctions", ""), JsonUtil.fromJson(
new TypeToken<List<FunctionDesp>>() {}.getType()); get.syncGet("", "getExportedFunctions", ""),
new TypeToken<List<FunctionDesp>>() {
}.getType());
contractMeta.logDetail = new HashMap<>(); contractMeta.logDetail = new HashMap<>();
for (FunctionDesp func : contractMeta.exportedFunctions) { for (FunctionDesp func : contractMeta.exportedFunctions) {
StringBuilder str = new StringBuilder(); StringBuilder str = new StringBuilder();
@ -159,7 +159,8 @@ public class ContractClient {
} }
if (anno.getType().equals("LogLocation")) { if (anno.getType().equals("LogLocation")) {
for (String logLoc : anno.getArgs()) { for (String logLoc : anno.getArgs()) {
if (logLoc.equals("\"dataware\"") || logLoc.equals("\"bdledger\"") if (logLoc.equals("\"dataware\"")
|| logLoc.equals("\"bdledger\"")
|| logLoc.equals("\"bdledger:\"")) { || logLoc.equals("\"bdledger:\"")) {
str.append("bdcontract;"); str.append("bdcontract;");
} else if (logLoc.startsWith("\"bdledger:") && logLoc.length() > 11) { } else if (logLoc.startsWith("\"bdledger:") && logLoc.length() > 11) {
@ -172,47 +173,54 @@ public class ContractClient {
} }
contractMeta.logDetail.put(func.functionName, str.toString()); contractMeta.logDetail.put(func.functionName, str.toString());
} }
LOGGER.info("initProps ---- position-----5"); // LOGGER.info("initProps ---- position-----5");
try { try {
String anno = get.syncGet("", "getAnnotations", ""); String anno = get.syncGet("", "getAnnotations", "");
contractMeta.annotations = contractMeta.annotations =
JsonUtil.fromJson(anno, new TypeToken<List<AnnotationNode>>() {}.getType()); JsonUtil.fromJson(anno, new TypeToken<List<AnnotationNode>>() {
}.getType());
} catch (Exception e) { } catch (Exception e) {
// supoort contract process before version 0.70 // supoort contract process before version 0.70
contractMeta.annotations = new ArrayList<>(); contractMeta.annotations = new ArrayList<>();
} }
LOGGER.info("initProps ---- position-----6"); // LOGGER.info("initProps ---- position-----6");
contractMeta.sigRequired = Boolean.parseBoolean(get.syncGet("", "isSigRequired", "")); contractMeta.sigRequired = Boolean.parseBoolean(get.syncGet("", "isSigRequired", ""));
LOGGER.info("initProps ---- position-----7"); // LOGGER.info("initProps ---- position-----7");
contractMeta.thisPermission = get.syncGet("", "showPermission", ""); contractMeta.thisPermission = get.syncGet("", "showPermission", "");
LOGGER.info("initProps ---- position-----8"); // LOGGER.info("initProps ---- position-----8");
isRunning = true; isRunning = true;
LOGGER.info("initProps ---- position-----9"); contractMeta.isDebug = Boolean.parseBoolean(get.syncGet("", "getDebug", ""));
// LOGGER.info("initProps ---- position-----9");
get.syncGet("", "registerMangerPort", ContractManager.cPort.getCMPort() + ""); get.syncGet("", "registerMangerPort", ContractManager.cPort.getCMPort() + "");
contractMeta.contract = contractMeta.contract =
JsonUtil.fromJson(get.syncGet("", "getContract", ""), Contract.class); JsonUtil.fromJson(get.syncGet("", "getContract", ""), Contract.class);
contractMeta.id = contractMeta.contract.getID(); contractMeta.id = contractMeta.contract.getID();
get.setOfflineExceptionHandler(new SocketGet.OfflineHandler() { get.setOfflineExceptionHandler(
@Override new SocketGet.OfflineHandler() {
public void onException(SocketGet socketGet, Exception e) { @Override
if (e.getMessage().contains("Connection refused")) { public void onException(SocketGet socketGet, Exception e) {
contractMeta.status = ContractStatusEnum.HANGED; if (e.getMessage().contains("Connection refused")) {
ContractManager.instance.statusRecorder.resumeContractProcess(contractMeta.id); contractMeta.status = ContractStatusEnum.HANGED;
} ContractManager.instance.statusRecorder.resumeContractProcess(
} contractMeta.id);
}); }
}
});
loadTimesAndTraffic(); loadTimesAndTraffic();
LOGGER.info("initProps ---- position-----10 DONE!"); // LOGGER.info("initProps ---- position-----10");
// LOGGER.debug("======= registerPort:" + ret + "-->" + ContractManager.startPort);
} }
public String startProcess(PrintStream ps) throws Exception { public String startProcess(PrintStream ps) {
isRunning = false; isRunning = false;
port = -1; port = -1;
String darg = "-Djava.library.path="; String darg = "-Djava.library.path=";
String classpath; String classpath;
File jniPath; File jniPath;
if (ContractManager.yjsPath == null) { if (ContractManager.yjsPath == null) {
jniPath = new File("./jni/"); jniPath = new File("./jni/");
classpath = System.getProperty("java.class.path"); classpath = System.getProperty("java.class.path");
@ -220,137 +228,122 @@ public class ContractClient {
classpath = ContractManager.yjsPath; classpath = ContractManager.yjsPath;
jniPath = new File(classpath).getParentFile(); jniPath = new File(classpath).getParentFile();
} }
String osJni = ((HardwareInfo.type == OSType.linux) ? "/jni/linux" : "/jni/mac"); String osJni = ((HardwareInfo.type == OSType.linux) ? "/jni/linux" : "/jni/mac");
darg += jniPath.getAbsolutePath() + osJni; darg += jniPath.getAbsolutePath() + osJni;
if (!new File(classpath).exists()) { if (!new File(classpath).exists()) {
ContractResult r = new ContractResult(Status.Exception, ContractResult r =
new JsonPrimitive("incorrect path: yjs.jar")); new ContractResult(
Status.Exception, new JsonPrimitive("incorrect path: yjs.jar"));
return JsonUtil.toJson(r); return JsonUtil.toJson(r);
} }
if (!new File(jniPath, "libs").exists()) { if (!new File(jniPath, "libs").exists()) {
ContractResult r = new ContractResult(Status.Exception, ContractResult r =
new JsonPrimitive("incorrect path: yjs.jar, missing libs")); new ContractResult(
Status.Exception,
new JsonPrimitive("incorrect path: yjs.jar, missing libs"));
return JsonUtil.toJson(r); return JsonUtil.toJson(r);
} }
// ProcessBuilder builder =
// new ProcessBuilder(
// "java",
// "-Dfile.encoding=UTF-8",
// darg,
// "-cp",
// jniPath.getAbsolutePath() + "/libs/*:" + classpath,
// "org.bdware.sc.ContractProcess",
// "-port=" + cPort.getPort());
int startPort = ContractManager.cPort.getPortAndInc(); int startPort = ContractManager.cPort.getPortAndInc();
List<String> pbParameters = new ArrayList<>(); ProcessBuilder builder =
pbParameters.add("java"); new ProcessBuilder(
pbParameters.add("-Dfile.encoding=UTF-8"); "java",
pbParameters.add(darg); "-Dfile.encoding=UTF-8",
if (contractMeta.contract.getRemoteDebugPort() != 0) { darg,
pbParameters.add(String.format( "-jar",
"-agentlib:jdwp=transport=dt_socket,address=%d,server=y,suspend=n", classpath,
contractMeta.contract.getRemoteDebugPort())); "-port=" + startPort,
} "-cmi=" + cmi, // cmi 区分不同CM的cp
File classParent = new File(classpath).getParentFile(); (isDebug ? "-debug" : ""));
if (contractMeta.contract.isDebug()) { File directory = new File("");
pbParameters.add("-Dlog4j.configurationFile="
+ new File(classParent, "log4j2.debug.properties").getAbsolutePath());
} else {
pbParameters.add("-Dlog4j.configurationFile="
+ new File(classParent, "log4j2.properties").getAbsolutePath());
}
pbParameters.add("-jar");
pbParameters.add(classpath);
pbParameters.add("-port=" + startPort);
pbParameters.add("-cmi=" + cmi);
if (contractMeta.contract.isDebug())
pbParameters.add("-debug");
ProcessBuilder builder;
String[] result = new String[pbParameters.size()];
pbParameters.toArray(result);
Constructor<ProcessBuilder> pbc =
ProcessBuilder.class.getDeclaredConstructor(String[].class);
builder = pbc.newInstance(new Object[] {result});
File directory = new File("./");
LOGGER.debug("[CMD] path: " + directory.getAbsolutePath()); LOGGER.debug("[CMD] path: " + directory.getAbsolutePath());
LOGGER.debug(JsonUtil.toPrettyJson(builder.command())); LOGGER.debug(JsonUtil.toPrettyJson(builder.command()));
Map<String, String> map = builder.environment(); Map<String, String> map = builder.environment();
map.put("java.library.path", jniPath.getAbsolutePath() + osJni); map.put("java.library.path", jniPath.getAbsolutePath() + osJni);
builder.directory(directory); builder.directory(new File("./"));
LOGGER.debug("start process:"); LOGGER.debug("start process:");
process = builder.start();
this.pid = getPid(process); try {
LOGGER.info("[CP PPID] " + pid); process = builder.start();
PrintStream printStream = new PrintStream(process.getOutputStream());
printStream.println("CP PID:" + pid); this.pid = getPid(process);
printStream.close(); LOGGER.info("[CP PPID ] " + pid);
InputStream processInputStream = process.getInputStream(); PrintStream printStream = new PrintStream(process.getOutputStream());
Scanner sc = new Scanner(processInputStream); printStream.println("CP PID:" + pid);
String status = null; printStream.close();
while (sc.hasNext()) { Scanner sc = new Scanner(process.getInputStream());
status = sc.nextLine(); String status = null;
LOGGER.info("[CP] " + status); while (sc.hasNext()) {
if (status.contains("mainPort")) { status = sc.nextLine();
try { LOGGER.info("[CP] " + status);
// Set contractPort to max(mainPort, contractPort) if (status.contains("mainPort")) {
int portIndex = status.indexOf("mainPort"); try {
int port = Integer.parseInt( // Set contractPort to max(mainPort, contractPort)
status.substring(portIndex + 9, portIndex + 14).replaceAll("\\s+", "")); int portIndex = status.indexOf("mainPort");
if (port != startPort) { int port =
ContractManager.cPort.reSetPort(port + 1); Integer.parseInt(
status.substring(portIndex + 9, portIndex + 14)
.replaceAll("\\s+", ""));
if (port != startPort) {
ContractManager.cPort.reSetPort(port + 1);
}
} catch (Exception ignored) {
} }
} catch (Exception ignored) { break;
} }
break;
} }
} if (status != null) {
if (status != null) { status = status.replaceAll(".*mainPort ", "");
status = status.replaceAll(".*mainPort ", ""); }
} assert status != null;
assert status != null; port = Integer.parseInt(status.split(" ")[0]);
port = Integer.parseInt(status.split(" ")[0]); LOGGER.debug("port=" + port);
LOGGER.debug("port=" + port); ContractManager.cPort.updateDb(port, true);
ContractManager.cPort.updateDb(port, true); get = new SocketGet("127.0.0.1", port);
get = new SocketGet("127.0.0.1", port); get.syncGet("", "setDBInfo", ContractManager.dbPath);
get.syncGet("", "setDBInfo", ContractManager.dbPath); String tagA = (ps == System.out ? "[Contract_" + port + "_out] " : "");
if (contractMeta.contract.isDebug() || get != null) { String tagB = (ps == System.out ? "[Contract_" + port + "_err] " : "");
String tagA = (ps instanceof LoggerPrintStream ? "[Contract_" + port + "_out] " : "");
String tagB = (ps instanceof LoggerPrintStream ? "[Contract_" + port + "_err] " : "");
outputTracer.track(process, sc, tagA, ps); outputTracer.track(process, sc, tagA, ps);
errorTracer.track(process, new Scanner(process.getErrorStream()), tagB, ps); errorTracer.track(process, new Scanner(process.getErrorStream()), tagB, ps);
} else { get.syncGet("", "registerMangerPort", String.valueOf(ContractManager.cPort.getCMPort()));
// 关闭流，否则缓冲区打满会阻塞进程
processInputStream.close();
sc.close();
}
get.syncGet("", "registerMangerPort", String.valueOf(ContractManager.cPort.getCMPort()));
MultiContractMeta multiContractMeta = ContractManager.instance.multiContractRecorder if (isBundlePath(contractMeta.contract.getScriptStr())) {
.getMultiContractMeta(contractMeta.getID()); status =
if (multiContractMeta != null && multiContractMeta.getMembers() != null) { get.syncGet(
String setMemberResult = "", "setContractBundle", JsonUtil.toJson(contractMeta.contract));
get.syncGet("", "setMembers", JsonUtil.toJson(multiContractMeta.getMembers())); } else {
LOGGER.info("setMember:" + setMemberResult); status = get.syncGet("", "setContract", JsonUtil.toJson(contractMeta.contract));
} else { }
LOGGER.info("setMember ignore, meta:" + (multiContractMeta == null) LOGGER.debug("port:" + port + " status:" + status);
+ (multiContractMeta == null ? "NULL" ContractResult r = JsonUtil.fromJson(status, ContractResult.class);
: " members:" + multiContractMeta.getMembers())); if (r.status == Status.Success) {
} initProps();
get.syncGet("", "setPID", pid);
if (isBundlePath(contractMeta.contract.getScriptStr())) { } else if (r.status == null) {
status = get.syncGet("", "setContractBundle", JsonUtil.toJson(contractMeta.contract));
} else {
status = get.syncGet("", "setContract", JsonUtil.toJson(contractMeta.contract));
}
LOGGER.info("start status, port:" + port + " status:" + status);
ContractResult r = JsonUtil.fromJson(status, ContractResult.class);
if (r.status == Status.Success) {
LOGGER.info("init prop start!");
initProps();
LOGGER.info("init prop done!");
get.syncGet("", "setPID", pid);
} else {
LOGGER.info("kill contract");
if (r.status == null) {
r.status = Status.Error; r.status = Status.Error;
r.result = new JsonPrimitive(status); r.result = new JsonPrimitive(status);
status = JsonUtil.toJson(r);
contractMeta.name = get.syncGet("", "getContractName", "");
} }
killProcess(); return status;
throw new IllegalStateException(r.result.getAsString()); } catch (Exception e) {
e.printStackTrace();
ByteArrayOutputStream bo = new ByteArrayOutputStream();
e.printStackTrace(new PrintStream(bo));
ContractResult r =
new ContractResult(Status.Exception, new JsonPrimitive(bo.toString()));
return JsonUtil.toJson(r);
} }
return status;
} }
public void updateMemory() { public void updateMemory() {
@ -359,12 +352,16 @@ public class ContractClient {
return; return;
} }
lastUpdate = System.currentTimeMillis(); lastUpdate = System.currentTimeMillis();
get.asyncGet("", ".UsedMemory", "", new ResultCallback() { get.asyncGet(
@Override "",
public void onResult(String str) { ".UsedMemory",
memory = Long.parseLong(str); "",
} new ResultCallback() {
}); @Override
public void onResult(String str) {
memory = Long.parseLong(str);
}
});
} catch (Exception e) { } catch (Exception e) {
// e.printStackTrace(); // e.printStackTrace();
} }
@ -378,8 +375,17 @@ public class ContractClient {
return contractMeta.getExportedFunctions(); return contractMeta.getExportedFunctions();
} }
public String executeMethod(String pkgName, String method, String arg) { public String getIdentifier() {
return get.syncGet(pkgName, method, arg); return get.syncGet("", "getIdentifier", " ");
}
public void setIdentifier(String alias) {
get.asyncGet("", "setIdentifier", alias, null);
}
public String signResult(String result) {
return contractMeta.contract.signResult(result);
} }
public String getPubkey() { public String getPubkey() {
@ -394,6 +400,10 @@ public class ContractClient {
return contractMeta.name; return contractMeta.name;
} }
public String getContractDOI() {
return contractMeta.contract.getDOI();
}
public String getContractKey() { public String getContractKey() {
return contractMeta.contract.getKey(); return contractMeta.contract.getKey();
} }
@ -418,6 +428,10 @@ public class ContractClient {
return contractMeta.logDetail.get(action); return contractMeta.logDetail.get(action);
} }
public boolean isDebug() {
return contractMeta.isDebug;
}
public long getTimes() { public long getTimes() {
return times; return times;
} }
@ -443,10 +457,12 @@ public class ContractClient {
} }
public void loadTimesAndTraffic() { public void loadTimesAndTraffic() {
String tempTime2 = KeyValueDBUtil.instance.getValue(CMTables.ContractInfo.toString(), String tempTime2 =
contractMeta.name + "-Times"); KeyValueDBUtil.instance.getValue(
String tempTraffic2 = KeyValueDBUtil.instance.getValue(CMTables.ContractInfo.toString(), CMTables.ContractInfo.toString(), contractMeta.name + "-Times");
contractMeta.name + "-Traffic"); String tempTraffic2 =
KeyValueDBUtil.instance.getValue(
CMTables.ContractInfo.toString(), contractMeta.name + "-Traffic");
if (tempTime2 != null && !tempTime2.equals("")) { if (tempTime2 != null && !tempTime2.equals("")) {
times = Long.parseLong(tempTime2); times = Long.parseLong(tempTime2);
} }
@ -456,15 +472,15 @@ public class ContractClient {
} }
public void saveTimesAndTraffic() { public void saveTimesAndTraffic() {
KeyValueDBUtil.instance.setValue(CMTables.ContractInfo.toString(), KeyValueDBUtil.instance.setValue(
contractMeta.name + "-Times", times + ""); CMTables.ContractInfo.toString(), contractMeta.name + "-Times", times + "");
KeyValueDBUtil.instance.setValue(CMTables.ContractInfo.toString(), KeyValueDBUtil.instance.setValue(
contractMeta.name + "-Traffic", traffic + ""); CMTables.ContractInfo.toString(), contractMeta.name + "-Traffic", traffic + "");
} }
public void setMask(JsonObject args) { public void setMask(JsonObject args) {
// get.asyncGet("",,,,); //get.asyncGet("",,,,);
get.asyncGet("", "setMask", args.toString(), null); get.asyncGet("", "setMask", args.toString(), null);
} }
@ -472,7 +488,7 @@ public class ContractClient {
public void setProjectConfig(String args) { public void setProjectConfig(String args) {
get.asyncGet("", "setProjectConfig", args, null); get.asyncGet("", "setProjectConfig", args, null);
} }
// public String //public String
static class ReqScript { static class ReqScript {
String mode; String mode;

1423
src/main/java/org/bdware/sc/ContractManager.java
View File

File diff suppressed because it is too large Load Diff

22
src/main/java/org/bdware/sc/ContractMeta.java
View File

@ -1,7 +1,5 @@
package org.bdware.sc; package org.bdware.sc;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.bdware.sc.bean.Contract; import org.bdware.sc.bean.Contract;
import org.bdware.sc.bean.FunctionDesp; import org.bdware.sc.bean.FunctionDesp;
import org.bdware.sc.bean.IDSerializable; import org.bdware.sc.bean.IDSerializable;
@ -22,6 +20,7 @@ public class ContractMeta implements IDSerializable {
ContractStatusEnum status; ContractStatusEnum status;
String name; String name;
String id; String id;
boolean isDebug;
Map<String, REvent.REventSemantics> declaredEvents; Map<String, REvent.REventSemantics> declaredEvents;
List<FunctionDesp> exportedFunctions; List<FunctionDesp> exportedFunctions;
Map<String, String> logDetail; Map<String, String> logDetail;
@ -31,8 +30,15 @@ public class ContractMeta implements IDSerializable {
boolean sigRequired; boolean sigRequired;
String thisPermission; // 合约当前权限 String thisPermission; // 合约当前权限
/* /*
* { "name": "dx_substr", "parameter": { "columnIndex":5, "paras":["1","3"] } }, {
*/ "name": "dx_substr",
"parameter":
{
"columnIndex":5,
"paras":["1","3"]
}
},
*/
// Map<Object,Object>MaskInfo; // Map<Object,Object>MaskInfo;
public ContractMeta() { public ContractMeta() {
@ -90,7 +96,7 @@ public class ContractMeta implements IDSerializable {
// public setMask(){} // public setMask(){}
public boolean getIsDebug() { public boolean getIsDebug() {
return contract.isDebug(); return isDebug;
} }
public FunctionDesp getExportedFunction(String action) { public FunctionDesp getExportedFunction(String action) {
@ -109,8 +115,7 @@ public class ContractMeta implements IDSerializable {
private FunctionDesp seekFunction(String action) { private FunctionDesp seekFunction(String action) {
for (FunctionDesp desp : exportedFunctions) { for (FunctionDesp desp : exportedFunctions) {
if (desp != null && desp.functionName.equals(action)) if (desp != null && desp.functionName.equals(action)) return desp;
return desp;
} }
return null; return null;
} }
@ -119,10 +124,9 @@ public class ContractMeta implements IDSerializable {
contract = c; contract = c;
id = c.getID(); id = c.getID();
status = ContractStatusEnum.HANGED; status = ContractStatusEnum.HANGED;
isDebug = false;
} }
static Logger LOGGER = LogManager.getLogger(ContractMeta.class);
public void setContractExecutor(ContractExecutor executor) { public void setContractExecutor(ContractExecutor executor) {
this.contractExecutor = executor; this.contractExecutor = executor;
} }

6
src/main/java/org/bdware/sc/ContractStatusEnum.java
View File

@ -1,8 +1,10 @@
package org.bdware.sc; package org.bdware.sc;
public enum ContractStatusEnum { public enum ContractStatusEnum {
INIT(0, "初始化"), RUNNING(1, "内存运行中"), HANGED(2, "硬盘运行中"), KILLED(3, "已停止"); INIT(0, "初始化"),
RUNNING(1, "内存运行中"),
HANGED(2, "硬盘运行中"),
KILLED(3, "已停止");
private Integer code; private Integer code;
private String desc; private String desc;

194
src/main/java/org/bdware/sc/ContractStatusRecorder.java
View File

@ -27,24 +27,28 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
// 对外会有一个getContractClient的过程，如果只是想用元信息的，改成getContractClientMeta // 对外会有一个getContractClient的过程，如果只是想用元信息的，改成getContractClientMeta
// 如果是想调用的，再使用getContractClient。 // 如果是想调用的，再使用getContractClient。
// 合约执行的状态 key:合同id value:合同状态 // 合约执行的状态 key:合同id value:合同状态
// private static final String DB_NAME = CMTables.ContractInfo.toString(); // private static final String DB_NAME = CMTables.ContractInfo.toString();
static { static {
final Object flag = new Object(); final Object flag = new Object();
// 调度一个task，在delay（ms）后开始调度，每次调度完后，最少等待period（ms）后才开始调 // 调度一个task，在delay（ms）后开始调度，每次调度完后，最少等待period（ms）后才开始调
ContractManager.scheduledThreadPool.scheduleWithFixedDelay(() -> { ContractManager.scheduledThreadPool.scheduleWithFixedDelay(
boolean cleared = dealTimerContractProcess(); () -> {
if (cleared) { boolean cleared = dealTimerContractProcess();
synchronized (flag) { if (cleared) {
try { synchronized (flag) {
flag.wait(14000L); try {
} catch (InterruptedException e) { flag.wait(14000L);
e.printStackTrace(); } catch (InterruptedException e) {
e.printStackTrace();
}
}
} }
} },
} 0,
}, 0, 1, TimeUnit.SECONDS); 1,
TimeUnit.SECONDS);
} }
// contract id to client, changes when some contract is started/stopped/resumed // contract id to client, changes when some contract is started/stopped/resumed
@ -61,32 +65,32 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
} }
public static boolean dealTimerContractProcess() { public static boolean dealTimerContractProcess() {
// if (id2ContractStatus == null) { // if (id2ContractStatus == null) {
// String value = readDataFromDB(id2ContractStatusKey); // String value = readDataFromDB(id2ContractStatusKey);
// id2ContractStatus = GSON.fromJson(value, Map.class); // id2ContractStatus = GSON.fromJson(value, Map.class);
// if (id2ContractStatus == null) { // if (id2ContractStatus == null) {
// return true; // return true;
// } // }
// for (Map.Entry<String, ContractStatusEnum> contractStatusEnumEntry : // for (Map.Entry<String, ContractStatusEnum> contractStatusEnumEntry :
// id2ContractStatus.entrySet()) { // id2ContractStatus.entrySet()) {
// //获取合同id // //获取合同id
// String contractID = contractStatusEnumEntry.getKey(); // String contractID = contractStatusEnumEntry.getKey();
// ContractStatusEnum contractStatusEnum = // ContractStatusEnum contractStatusEnum =
// contractStatusEnumEntry.getValue(); // contractStatusEnumEntry.getValue();
// //没有状态信息就停止 // //没有状态信息就停止
// if (contractStatusEnum == null) { // if (contractStatusEnum == null) {
// continue; // continue;
// } // }
// //拥有合同状态信息的，如果是已经执行结束，我们将合同挂起 // //拥有合同状态信息的，如果是已经执行结束，我们将合同挂起
// if // if
// (contractStatusEnum.getCode().equals(ContractStatusEnum.STORE.getCode())) { // (contractStatusEnum.getCode().equals(ContractStatusEnum.STORE.getCode())) {
// resumeContractProcess(contractID); // resumeContractProcess(contractID);
// } else if // } else if
// (contractStatusEnum.getCode().equals(ContractStatusEnum.DONE.getCode())) { // (contractStatusEnum.getCode().equals(ContractStatusEnum.DONE.getCode())) {
// hangUpContractProcess(contractID); // hangUpContractProcess(contractID);
// } // }
// } // }
// } // }
return true; return true;
} }
@ -98,11 +102,10 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
// TODO 是不是还有别的？这只是在内存里的哦。 // TODO 是不是还有别的？这只是在内存里的哦。
try { try {
for (ContractClient c : id2ContractClient.values()) { for (ContractClient c : id2ContractClient.values()) {
if (Integer.parseInt(c.getPID()) == pid) if (Integer.parseInt(c.getPID()) == pid) return true;
return true;
} }
} catch (Exception e) { } catch (Exception e) {
// e.printStackTrace(); // e.printStackTrace();
} }
return false; return false;
} }
@ -171,23 +174,17 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
// TODO index name to contract // TODO index name to contract
public ContractMeta getContractMeta(String idOrNameOrDOI) { public ContractMeta getContractMeta(String idOrNameOrDOI) {
if (idOrNameOrDOI == null) if (idOrNameOrDOI == null) return null;
return null;
ContractMeta meta = getStatus().get(idOrNameOrDOI); ContractMeta meta = getStatus().get(idOrNameOrDOI);
if (meta != null) if (meta != null) return meta;
return meta;
for (ContractMeta cc : getStatus().values()) { for (ContractMeta cc : getStatus().values()) {
if (cc.status == ContractStatusEnum.RUNNING || cc.status == ContractStatusEnum.HANGED) if (cc.status == ContractStatusEnum.RUNNING || cc.status == ContractStatusEnum.HANGED)
if (idOrNameOrDOI.equals(cc.name)) { if (idOrNameOrDOI.equals(cc.name) || idOrNameOrDOI.equals(cc.contract.getDOI())) {
return cc; return cc;
} }
} }
for (ContractMeta cc : getStatus().values()) { for (ContractMeta cc : getStatus().values()) {
if (cc.name == null) if (idOrNameOrDOI.equals(cc.name) || idOrNameOrDOI.equals(cc.contract.getDOI())) {
continue;
if (cc.contract == null)
continue;
if (idOrNameOrDOI.equals(cc.name)) {
return cc; return cc;
} }
} }
@ -225,8 +222,10 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
LOGGER.info("need dump when stop"); LOGGER.info("need dump when stop");
String contractName = client.getContractName(); String contractName = client.getContractName();
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd.HH:mm:ss"); SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd.HH:mm:ss");
File f = new File(ContractManager.dir + "/memory/" + contractName, File f =
df.format(new Date())); new File(
ContractManager.dir + "/memory/" + contractName,
df.format(new Date()));
client.get.syncGet("", "getMemoryDump", f.getAbsolutePath()); client.get.syncGet("", "getMemoryDump", f.getAbsolutePath());
ContractManager.instance.addLocalContractLog("dumpContract", meta); ContractManager.instance.addLocalContractLog("dumpContract", meta);
} }
@ -280,9 +279,9 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
resumeContractProcess(contractID); resumeContractProcess(contractID);
} }
// 似乎这样就可以了？ // 似乎这样就可以了？
// } else if (meta.status == ContractStatusEnum.RUNNING) { // } else if (meta.status == ContractStatusEnum.RUNNING) {
// hangUpContractProcess(contractID); // hangUpContractProcess(contractID);
// } // }
} }
/** /**
@ -290,15 +289,15 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
*/ */
public synchronized void hangUpContractProcess(String contractID) { public synchronized void hangUpContractProcess(String contractID) {
// dump stop 这里是正常流程代码 // dump stop 这里是正常流程代码
// ContractClient contractClient = getContractClientById(contractID); // ContractClient contractClient = getContractClientById(contractID);
// if (contractClient == null) { // if (contractClient == null) {
// return; // return;
// } // }
// //挂起 // //挂起
// Contract contract = contractClient.contract; // Contract contract = contractClient.contract;
// ContractManager.instance.dumpContract(contract.getID(), null); // ContractManager.instance.dumpContract(contract.getID(), null);
// setContractClient(contractID, contractClient); // setContractClient(contractID, contractClient);
// ContractManager.instance.hangUpStopContract(contract.getID()); // ContractManager.instance.hangUpStopContract(contract.getID());
// todo 临时想强制测试下效果 // todo 临时想强制测试下效果
ContractMeta contractMeta = getContractMeta(contractID); ContractMeta contractMeta = getContractMeta(contractID);
// 挂起 // 挂起
@ -312,12 +311,13 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
client.saveTimesAndTraffic(); client.saveTimesAndTraffic();
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd.HH_mm_ss"); // 设置日期格式 SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd.HH_mm_ss"); // 设置日期格式
File f = new File(ContractManager.dir + "/memory/" + contractMeta.name, File f =
df.format(new Date())); new File(
ContractManager.dir + "/memory/" + contractMeta.name,
df.format(new Date()));
File parent = f.getParentFile(); File parent = f.getParentFile();
if (!parent.exists()) { if (!parent.exists()) {
LOGGER.trace( LOGGER.trace("make directory " + parent.getAbsolutePath() + ":" + parent.mkdirs());
"make directory " + parent.getAbsolutePath() + ":" + parent.mkdirs());
} }
ContractManager.instance.dumpContract(contract.getID(), f.getAbsolutePath()); ContractManager.instance.dumpContract(contract.getID(), f.getAbsolutePath());
ContractManager.instance.invokeContractSuicide(client); ContractManager.instance.invokeContractSuicide(client);
@ -333,7 +333,7 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
// 启动合约 // 启动合约
// 加载所需要的内存 // 加载所需要的内存
// 该方法调用方法已经考虑了内存问题，但是内存不足的时候，重试或者等待，或者将合约压到失败中稍后重试，可以尝试设计消息重试机制 // 该方法调用方法已经考虑了内存问题，但是内存不足的时候，重试或者等待，或者将合约压到失败中稍后重试，可以尝试设计消息重试机制
// 如果是内存中则唤起合同信息 todo 内存不足 关闭合约追赶，补偿数据 // 如果是内存中则唤起合同信息 todo 内存不足 关闭合约追赶，补偿数据
if (meta.status == ContractStatusEnum.HANGED || meta.status == ContractStatusEnum.KILLED) { if (meta.status == ContractStatusEnum.HANGED || meta.status == ContractStatusEnum.KILLED) {
ContractStatusEnum preStatus = meta.status; ContractStatusEnum preStatus = meta.status;
// TODO Resume里复用的是start逻辑，启动完了就成Running了哦。可能不能这么来。 // TODO Resume里复用的是start逻辑，启动完了就成Running了哦。可能不能这么来。
@ -342,25 +342,33 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
JsonUtil.fromJson(startContractResult, ContractResult.class); JsonUtil.fromJson(startContractResult, ContractResult.class);
if (contractResult.status == ContractResult.Status.Error) { if (contractResult.status == ContractResult.Status.Error) {
// 记录错误日志 // 记录错误日志
ContractManager.instance.addLocalContractLog("resumeContract error", ContractManager.instance.addLocalContractLog(
meta.contract.getID(), meta.name, meta.contract.getOwner()); "resumeContract error",
meta.contract.getID(),
meta.name,
meta.contract.getOwner());
} else { } else {
ContractManager.instance.addLocalContractLog("resumeContract success", ContractManager.instance.addLocalContractLog(
meta.contract.getID(), meta.name, meta.contract.getOwner()); "resumeContract success",
meta.contract.getID(),
meta.name,
meta.contract.getOwner());
// TODO 可能会重复load相同的镜像？ // TODO 可能会重复load相同的镜像？
// 如果是killed 只根据manifest去判断是否加载memory // 如果是killed 只根据manifest去判断是否加载memory
// 可增加一个判断，如果hanged朋manifest里是加载memory，这里就不再加载了。 // 可增加一个判断，如果hanged朋manifest里是加载memory，这里就不再加载了。
ContractClient client = id2ContractClient.get(meta.getID()); ContractClient client = id2ContractClient.get(meta.getID());
if (preStatus == ContractStatusEnum.HANGED) { if (preStatus == ContractStatusEnum.HANGED) {
String memory = ContractManager.instance.findNewestMemory(meta.name);
if (memory != null) ContractManager.instance.loadMemory(
ContractManager.instance.loadMemory(client, memory); client, ContractManager.instance.findNewestMemory(meta.name));
} }
client.contractMeta.setStatus(ContractStatusEnum.RUNNING); client.contractMeta.setStatus(ContractStatusEnum.RUNNING);
updateValue(client.contractMeta); updateValue(client.contractMeta);
// 还要判断一发，是不是已有permission的值。如果有，要setPermission一把。 // 还要判断一发，是不是已有permission的值。如果有，要setPermission一把。
// meta.status = ContractStatusEnum.RUNNING; // meta.status = ContractStatusEnum.RUNNING;
// updateValue(meta); // updateValue(meta);
} }
} }
} }
@ -381,11 +389,6 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
@Override @Override
public boolean visit(int i) { public boolean visit(int i) {
return visit("127.0.0.1", i);
}
@Override
public boolean visit(String host, int i) {
if (i == ContractManager.cPort.getCMPort()) { if (i == ContractManager.cPort.getCMPort()) {
return false; return false;
} }
@ -393,7 +396,7 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
return true; return true;
} }
try { try {
ContractClient client = new ContractClient(host, i); ContractClient client = new ContractClient(i);
// LOGGER.info("[CM重连] position-----1"); // LOGGER.info("[CM重连] position-----1");
if (client.isRunning) { if (client.isRunning) {
LOGGER.debug("CP listened to port " + i + " is running"); LOGGER.debug("CP listened to port " + i + " is running");
@ -408,23 +411,30 @@ public class ContractStatusRecorder extends StatusRecorder<ContractMeta> {
ContractManager.instance.invokeContractSuicide(client); ContractManager.instance.invokeContractSuicide(client);
return false; return false;
} }
LOGGER.debug(String.format( LOGGER.debug(
"CP listened to port %d:\n\tID=%s\n\tName=%s\n\tType=%s\n" String.format(
+ "\tKey=%s\n\tPubKey=%s\n\tCopies=%d\n\t%s", "CP listened to port %d:\n\tID=%s\n\tName=%s\n\tType=%s\n"
i, client.getContractID(), client.contractMeta.name, + "\tKey=%s\n\tPubKey=%s\n\tCopies=%d\n\t%s",
client.getContractType(), client.getContractKey(), client.getPubkey(), i,
client.getContractCopies(), client.contractMeta.contract.startInfo)); client.getContractID(),
client.contractMeta.name,
client.getContractType(),
client.getContractKey(),
client.getPubkey(),
client.getContractCopies(),
client.contractMeta.contract.startInfo));
client.get.syncGet("", "setDir", ContractManager.dir); client.get.syncGet("", "setDir", ContractManager.dir);
// String str = client.getIdentifier(); // String str = client.getIdentifier();
client.get.syncGet("", "getDumpPeriod", "a"); client.get.syncGet("", "getDumpPeriod", "a");
client.get.syncGet("", "startAutoDump", "a"); client.get.syncGet("", "startAutoDump", "a");
createContract(client); createContract(client);
LOGGER.info(String.format("reconnect to port %d: contract %s", i, LOGGER.info(
client.contractMeta.name)); String.format("reconnect to port %d: contract %s",
i, client.contractMeta.name));
return true; return true;
} }
} catch (Exception e) { } catch (Exception e) {
// just ignore //just ignore
} }
return false; return false;
} }

19
src/main/java/org/bdware/sc/LoggerPrintStream.java
View File

@ -1,19 +0,0 @@
package org.bdware.sc;
import org.apache.logging.log4j.Logger;
import java.io.PrintStream;
public class LoggerPrintStream extends PrintStream {
public LoggerPrintStream(Logger logger) {
super(System.out, true);
this.logger = logger;
}
Logger logger;
@Override
public void println(String str) {
logger.info(str);
}
}

16
src/main/java/org/bdware/sc/MasterElectTimeRecorder.java
View File

@ -1,13 +1,15 @@
package org.bdware.sc; package org.bdware.sc;
import java.util.HashMap;
import java.util.Map;
// Node //Node
public class MasterElectTimeRecorder { public class MasterElectTimeRecorder {
public static Long findMasterCrash; // 发现master崩溃的时间 public static Long findMasterCrash; //发现master崩溃的时间
public static Long newMasterStart; // 作为新的master newMasterStart被调用时间 public static Long newMasterStart; //作为新的master newMasterStart被调用时间
public static Long setLocalMaster; // 在本地标记自己为master public static Long setLocalMaster; //在本地标记自己为master
public static Long slaveConnectFinish; // 被slave连接完成 public static Long slaveConnectFinish; //被slave连接完成
public static Long masterStartRecover; // 开始master恢复 public static Long masterStartRecover; //开始master恢复
public static Long masterRecoverFinish; // master恢复结束 public static Long masterRecoverFinish; //master恢复结束
} }

34
src/main/java/org/bdware/sc/MultiContractRecorder.java
View File

@ -7,9 +7,6 @@ import org.bdware.sc.db.KeyValueDBUtil;
import org.bdware.sc.db.StatusRecorder; import org.bdware.sc.db.StatusRecorder;
import org.bdware.sc.units.MultiContractMeta; import org.bdware.sc.units.MultiContractMeta;
import java.util.HashSet;
import java.util.Set;
public class MultiContractRecorder extends StatusRecorder<MultiContractMeta> { public class MultiContractRecorder extends StatusRecorder<MultiContractMeta> {
static final String dbName = CMTables.UnitContracts.toString(); static final String dbName = CMTables.UnitContracts.toString();
static final String prefix = "Multi_C_Meta_"; static final String prefix = "Multi_C_Meta_";
@ -20,31 +17,26 @@ public class MultiContractRecorder extends StatusRecorder<MultiContractMeta> {
for (MultiContractMeta meta : getStatus().values()) { for (MultiContractMeta meta : getStatus().values()) {
try { try {
meta.initQueue(); meta.initQueue();
int lastExeSeq = Integer.parseInt(KeyValueDBUtil.instance int lastExeSeq =
.getValue(CMTables.LastExeSeq.toString(), meta.getContractID())); Integer.parseInt(
KeyValueDBUtil.instance.getValue(
CMTables.LastExeSeq.toString(), meta.getContractID()));
meta.setLastExeSeq(lastExeSeq); meta.setLastExeSeq(lastExeSeq);
} catch (Exception e) { } catch (Exception e) {
LOGGER.error(e); e.printStackTrace();
} }
} }
} }
public MultiContractMeta getMultiContractMeta(String idOrNameOrDOI) { public MultiContractMeta getMultiContractMeta(String idOrNameOrDOI) {
if (idOrNameOrDOI == null) if (idOrNameOrDOI == null) return null;
return null;
ContractMeta meta = ContractManager.instance.statusRecorder.getContractMeta(idOrNameOrDOI); ContractMeta meta = ContractManager.instance.statusRecorder.getContractMeta(idOrNameOrDOI);
return getMultiContractMeta(meta); if (meta == null) return null;
}
public MultiContractMeta getMultiContractMeta(ContractMeta meta) {
if (meta == null)
return null;
return getStatus().get(meta.id); return getStatus().get(meta.id);
} }
public MultiContractMeta createIfNotExist(String contractID) { public MultiContractMeta createIfNotExist(String contractID) {
ContractMeta meta = ContractManager.instance.statusRecorder.createIfNotExist(contractID); MultiContractMeta ret = getMultiContractMeta(contractID);
MultiContractMeta ret = getMultiContractMeta(meta);
if (null == ret) { if (null == ret) {
LOGGER.info("requests don't contain contract " + contractID); LOGGER.info("requests don't contain contract " + contractID);
ret = new MultiContractMeta(contractID); ret = new MultiContractMeta(contractID);
@ -52,14 +44,4 @@ public class MultiContractRecorder extends StatusRecorder<MultiContractMeta> {
} }
return ret; return ret;
} }
public void removeNotExist(ContractStatusRecorder statusRecorder) {
Set<MultiContractMeta> toRemove = new HashSet<>();
for (String key : getStatus().keySet()) {
if (statusRecorder.getContractMeta(key) == null)
toRemove.add(getStatus().get(key));
}
for (MultiContractMeta meta : toRemove)
remove(meta);
}
} }

2
src/main/java/org/bdware/sc/ProjectRecorder.java
View File

@ -7,7 +7,7 @@ import org.bdware.sc.db.StatusRecorder;
public class ProjectRecorder extends StatusRecorder<ProjectConfig> { public class ProjectRecorder extends StatusRecorder<ProjectConfig> {
static final String dbName = CMTables.ProjectConfig.toString(); static final String dbName = CMTables.ProjectConfig.toString();
static final String prefix = "Project_Config_"; static final String prefix = "Project_Config_";
// private static final Logger LOGGER = LogManager.getLogger(ProjectRecorder.class); // private static final Logger LOGGER = LogManager.getLogger(ProjectRecorder.class);
public ProjectRecorder(String dir) { public ProjectRecorder(String dir) {
super(dir, dbName, prefix); super(dir, dbName, prefix);

48
src/main/java/org/bdware/sc/RecoverMechTimeRecorder.java
View File

@ -4,41 +4,41 @@ import java.util.HashMap;
import java.util.Map; import java.util.Map;
public class RecoverMechTimeRecorder { public class RecoverMechTimeRecorder {
// slave记录 //slave记录
public static Long startCMHttpServer; // 启动 public static Long startCMHttpServer; //启动
public static Long startFinish; // 启动完成，开始连接NC public static Long startFinish; //启动完成，开始连接NC
public static Long connectNCFinish; // 连接NC完成 public static Long connectNCFinish; //连接NC完成
public static Long startReconnectCP; // 开始重连CP public static Long startReconnectCP; //开始重连CP
public static Long reconnectCPFinish; // 重连CP完成 public static Long reconnectCPFinish; //重连CP完成
public static Long startQueryMaster; // 开始查看master是谁 public static Long startQueryMaster; //开始查看master是谁
public static Long queryMasterFinish; // 查到master是谁，开始连接master public static Long queryMasterFinish; //查到master是谁，开始连接master
public static Long connectMasterFinish; // 连接master完成 public static Long connectMasterFinish; //连接master完成
public static Long startRecover; // 开始恢复，slave的askForRecover被调用 public static Long startRecover; //开始恢复，slave的askForRecover被调用
public static Long startCommonRecover; // slave开始从common恢复 public static Long startCommonRecover; //slave开始从common恢复
public static Long startStableRecover; // slave开始从stable恢复 public static Long startStableRecover; //slave开始从stable恢复
public static Long stableLoadFinish; // stable恢复模式load完成 public static Long stableLoadFinish; //stable恢复模式load完成
public static Long stableRedoFinish; // stable恢复模式redo完成 public static Long stableRedoFinish; //stable恢复模式redo完成
public static Long startRedoTransFromMaster; // redo从master传来的trans public static Long startRedoTransFromMaster; //redo从master传来的trans
// master记录,key是slave的pubKey //master记录,key是slave的pubKey
public static Map<String, Long> masterStartRecoverNode = new HashMap<String, Long>(); // master的askForRecover被调用 public static Map<String,Long> masterStartRecoverNode = new HashMap<String, Long>(); //master的askForRecover被调用
public static Map<String, Long> startJudgeRecoverMethod = new HashMap<String, Long>(); // 开始判断某个节点的恢复方式 public static Map<String,Long> startJudgeRecoverMethod = new HashMap<String,Long>(); //开始判断某个节点的恢复方式
public static Map<String, Long> judgeRecoverMethodFinish = new HashMap<String, Long>(); // 判断某节点恢复方式完成 public static Map<String,Long> judgeRecoverMethodFinish = new HashMap<String,Long>(); //判断某节点恢复方式完成
public static Map<String, Long> startRecoverFromCommon = new HashMap<String, Long>(); // slave从common恢复master的nodeRestartFromCommonMode方法开始时间 public static Map<String,Long> startRecoverFromCommon = new HashMap<String,Long>(); //slave从common恢复master的nodeRestartFromCommonMode方法开始时间
public static Map<String, Long> writeCEIStart = new HashMap<String, Long>(); // 开始dumpContract等cei信息 public static Map<String,Long> writeCEIStart = new HashMap<String, Long>(); //开始dumpContract等cei信息
public static Map<String, Long> finishWriteCEI = new HashMap<String, Long>(); // 将cei写入文件完成 public static Map<String,Long> finishWriteCEI = new HashMap<String, Long>(); //将cei写入文件完成
public static Map<String, Long> startRecoverFromStable = new HashMap<String, Long>(); // slave从common恢复master的restartFromStableMode方法开始时间 public static Map<String,Long> startRecoverFromStable = new HashMap<String,Long>(); //slave从common恢复master的restartFromStableMode方法开始时间
public static Map<String, Long> recoverFinish = new HashMap<String, Long>(); public static Map<String,Long> recoverFinish = new HashMap<String,Long>();
} }

92
src/main/java/org/bdware/sc/YPKResourceManager.java
View File

@ -1,92 +0,0 @@
package org.bdware.sc;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.bdware.sc.bean.Contract;
import java.io.*;
import java.nio.file.Files;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
public class YPKResourceManager {
public static String ypkCacheDir = "./ypkcache";
static ExecutorService es = Executors.newFixedThreadPool(2);
public static void unzipYPK(Contract contract) {
es.execute(() -> unzip(contract.getScriptStr(), getUnzipDir(contract)));
}
public static String getUnzipDir(Contract contract) {
return ypkCacheDir + "/" + contract.getID();
}
private static final long MAX_CACHE_SIZE = 2L * 1024 * 1024 * 1024; // 2GB
private static final long MAX_FILE_SIZE = 10 * 1024 * 1024; // 10MB
private static final Cache<String, byte[]> cache = Caffeine.newBuilder()
.maximumSize(MAX_CACHE_SIZE).expireAfterAccess(30, TimeUnit.MINUTES).build();
public static InputStream getCachedStream(Contract c, String path) throws IOException {
File target = new File(getUnzipDir(c), path);
String key = target.getAbsolutePath();
if (target.length() > MAX_FILE_SIZE) {
// 文件过大，直接从文件系统读取
return new FileInputStream(target);
} else {
byte[] data = cache.get(key, k -> {
try {
return Files.readAllBytes(target.toPath());
} catch (IOException e) {
throw new RuntimeException(e);
}
});
// 从缓存中返回一个新的 ByteArrayInputStream
return new ByteArrayInputStream(data);
}
}
public static void unzip(String zipFilePath, String destDir) {
File dir = new File(destDir);
if (!dir.exists())
dir.mkdirs();
byte[] buffer = new byte[1024];
try {
ZipInputStream zis = new ZipInputStream(new FileInputStream(zipFilePath));
ZipEntry zipEntry = zis.getNextEntry();
while (zipEntry != null) {
String fileName = zipEntry.getName();
File newFile = new File(destDir + File.separator + fileName);
// 创建所有上级目录
new File(newFile.getParent()).mkdirs();
if (zipEntry.isDirectory()) {
if (!newFile.isDirectory() && !newFile.mkdirs()) {
throw new IOException("Failed to create directory " + newFile);
}
} else {
// 写入文件内容
FileOutputStream fos = new FileOutputStream(newFile);
int len;
while ((len = zis.read(buffer)) > 0) {
fos.write(buffer, 0, len);
}
fos.close();
}
zipEntry = zis.getNextEntry();
}
zis.closeEntry();
zis.close();
} catch (IOException e) {
e.printStackTrace();
}
}
public static boolean existYPK(Contract contract) {
return new File(contract.getScriptStr()).exists();
}
}

2
src/main/java/org/bdware/sc/consistency/CommitAlgorithm.java
View File

@ -9,4 +9,4 @@ public interface CommitAlgorithm {
void setCommitter(Committer c); void setCommitter(Committer c);
void setConnection(TrustfulExecutorConnection c); void setConnection(TrustfulExecutorConnection c);
} }

12
src/main/java/org/bdware/sc/consistency/Pair.java
View File

@ -1,11 +1,11 @@
package org.bdware.sc.consistency; package org.bdware.sc.consistency;
public class Pair<T1, T2> { public class Pair<T1, T2> {
public T1 first; public T1 first;
public T2 second; public T2 second;
public Pair(T1 t1, T2 t2) { public Pair(T1 t1, T2 t2) {
first = t1; first = t1;
second = t2; second = t2;
} }
} }

42
src/main/java/org/bdware/sc/consistency/pbft/PBFTAlgorithm.java
View File

@ -65,9 +65,9 @@ public class PBFTAlgorithm implements CommitAlgorithm {
LOGGER.info("recv: " + pbftMessage.getDisplayStr()); LOGGER.info("recv: " + pbftMessage.getDisplayStr());
switch (pbftMessage.type) { switch (pbftMessage.type) {
case AddMember: case AddMember:
// PBFTMember member = PBFTMember.parse(pbftMessage.content); // PBFTMember member = PBFTMember.parse(pbftMessage.content);
// if (member != null) // if (member != null)
// members.put(sender, member); // members.put(sender, member);
// case DeleteMember: // case DeleteMember:
// JUST ignore // JUST ignore
break; break;
@ -79,8 +79,7 @@ public class PBFTAlgorithm implements CommitAlgorithm {
prepareMsg = new PBFTMessage(); prepareMsg = new PBFTMessage();
prepareMsg.order = allocatedID.incrementAndGet(); prepareMsg.order = allocatedID.incrementAndGet();
prepareMsg.type = PBFTType.PrePrepare; prepareMsg.type = PBFTType.PrePrepare;
prepareMsg.content = prepareMsg.content = (java.util.Arrays.hashCode(pbftMessage.content) + "").getBytes();
(java.util.Arrays.hashCode(pbftMessage.content) + "").getBytes();
temp = new PCInfo(); temp = new PCInfo();
temp.request = pbftMessage; temp.request = pbftMessage;
@ -109,8 +108,7 @@ public class PBFTAlgorithm implements CommitAlgorithm {
break; break;
case Prepare: case Prepare:
temp = info.get(pbftMessage.order); temp = info.get(pbftMessage.order);
LOGGER.info( LOGGER.info("receive Prepare from:" + pbftMessage.sendID + " -> " + pbftMessage.order);
"receive Prepare from:" + pbftMessage.sendID + " -> " + pbftMessage.order);
if (temp == null) { if (temp == null) {
PCInfo pcInfo = new PCInfo(); PCInfo pcInfo = new PCInfo();
pcInfo.buff.add(pbftMessage); pcInfo.buff.add(pbftMessage);
@ -157,8 +155,7 @@ public class PBFTAlgorithm implements CommitAlgorithm {
prepareMsg.order = pbftMessage.order; prepareMsg.order = pbftMessage.order;
prepareMsg.type = PBFTType.PrePrepare; prepareMsg.type = PBFTType.PrePrepare;
temp = info.get(prepareMsg.order); temp = info.get(prepareMsg.order);
prepareMsg.content = prepareMsg.content = (java.util.Arrays.hashCode(temp.request.content) + "").getBytes();
(java.util.Arrays.hashCode(temp.request.content) + "").getBytes();
broadcast(pbftMessage); broadcast(pbftMessage);
} }
default: default:
@ -199,19 +196,18 @@ public class PBFTAlgorithm implements CommitAlgorithm {
private void requestPrePrepareFromMaster(long order) { private void requestPrePrepareFromMaster(long order) {
// PBFTMessage message = new PBFTMessage(); // PBFTMessage message = new PBFTMessage();
// message.type = PBFTType.ReSend; // message.type = PBFTType.ReSend;
// message.order = order; // message.order = order;
// message.content = new byte[1]; // message.content = new byte[1];
// sendToMaster(message); // sendToMaster(message);
LOGGER.info("request Resend"); LOGGER.info("request Resend");
} }
private void matchPrePrepareFromOriginReqeust(int hash, PBFTMessage pbftMessage) { private void matchPrePrepareFromOriginReqeust(int hash, PBFTMessage pbftMessage) {
for (PCInfo pcInfo : info.values()) { for (PCInfo pcInfo : info.values()) {
for (PBFTMessage msg : pcInfo.buff) { for (PBFTMessage msg : pcInfo.buff) {
if (msg.type == PBFTType.PrePrepare if (msg.type == PBFTType.PrePrepare && Integer.parseInt(new String(msg.content)) == hash) {
&& Integer.parseInt(new String(msg.content)) == hash) {
handlePrePrepare(msg, original.get(hash).second); handlePrePrepare(msg, original.get(hash).second);
return; return;
} }
@ -224,8 +220,10 @@ public class PBFTAlgorithm implements CommitAlgorithm {
} }
private void retryLater(int delay, final Node sender, final PBFTMessage pbftMessage) { private void retryLater(int delay, final Node sender, final PBFTMessage pbftMessage) {
ContractManager.scheduledThreadPool.schedule(() -> onPBFTMessage(sender, pbftMessage), ContractManager.scheduledThreadPool.schedule(
delay, TimeUnit.MILLISECONDS); () -> onPBFTMessage(sender, pbftMessage),
delay,
TimeUnit.MILLISECONDS);
} }
private void handlePrePrepare(PBFTMessage pbftMessage, PBFTMessage req) { private void handlePrePrepare(PBFTMessage pbftMessage, PBFTMessage req) {
@ -374,8 +372,8 @@ public class PBFTAlgorithm implements CommitAlgorithm {
if (!isPrePrepareReceived) { if (!isPrePrepareReceived) {
return false; return false;
} }
// System.out.println("---: updatePrepare, size:" + prepare.size() + " -->" // System.out.println("---: updatePrepare, size:" + prepare.size() + " -->"
// + (center.members.size() / 3 * 2 + 1) + " --senderID:" + message.sendID); // + (center.members.size() / 3 * 2 + 1) + " --senderID:" + message.sendID);
if (prepare.size() > center.members.size() / 3 * 2) { if (prepare.size() > center.members.size() / 3 * 2) {
isSendCommit = true; isSendCommit = true;
@ -401,8 +399,8 @@ public class PBFTAlgorithm implements CommitAlgorithm {
public String getDisplayStr() { public String getDisplayStr() {
return "pSize:" + (prepare == null ? "null" : prepare.size()) + " cSize:" return "pSize:" + (prepare == null ? "null" : prepare.size()) + " cSize:"
+ (commit == null ? "null" : commit.size()) + " isSendCommit:" + isSendCommit + (commit == null ? "null" : commit.size()) + " isSendCommit:" + isSendCommit + " isSendReply:"
+ " isSendReply:" + isSendReply + " buffSize:" + buff.size(); + isSendReply + " buffSize:" + buff.size();
} }
} }
} }

56
src/main/java/org/bdware/sc/consistency/pbft/PBFTMember.java
View File

@ -9,35 +9,35 @@ import java.io.Serializable;
public class PBFTMember implements Serializable { public class PBFTMember implements Serializable {
private static final long serialVersionUID = -3058672609865345062L; private static final long serialVersionUID = -3058672609865345062L;
public boolean isMaster; public boolean isMaster;
public boolean isMaster() { public boolean isMaster() {
return isMaster; return isMaster;
} }
public static PBFTMember parse(byte[] content) { public static PBFTMember parse(byte[] content) {
try { try {
ByteArrayInputStream input = new ByteArrayInputStream(content); ByteArrayInputStream input = new ByteArrayInputStream(content);
ObjectInputStream objectInputStream = new ObjectInputStream(input); ObjectInputStream objectInputStream = new ObjectInputStream(input);
PBFTMember ret = (PBFTMember) objectInputStream.readObject(); PBFTMember ret = (PBFTMember) objectInputStream.readObject();
return ret; return ret;
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
return null; return null;
} }
public byte[] toByteArray() { public byte[] toByteArray() {
ObjectOutputStream out; ObjectOutputStream out;
try { try {
ByteArrayOutputStream bo = new ByteArrayOutputStream(); ByteArrayOutputStream bo = new ByteArrayOutputStream();
out = new ObjectOutputStream(bo); out = new ObjectOutputStream(bo);
out.writeObject(this); out.writeObject(this);
return bo.toByteArray(); return bo.toByteArray();
} catch (IOException e) { } catch (IOException e) {
e.printStackTrace(); e.printStackTrace();
} }
return null; return null;
} }
} }

166
src/main/java/org/bdware/sc/consistency/pbft/PBFTMessage.java
View File

@ -7,97 +7,97 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException; import java.io.IOException;
public class PBFTMessage { public class PBFTMessage {
PBFTType type; PBFTType type;
int sendID; int sendID;
long order; long order;
byte[] content; byte[] content;
String stamp;// time-stamp or other string, given by client, to identify the content String stamp;// time-stamp or other string, given by client, to identify the content
String signature; String signature;
public int getBytesHash() { public int getBytesHash() {
ByteArrayOutputStream bo = new ByteArrayOutputStream(); ByteArrayOutputStream bo = new ByteArrayOutputStream();
bo.write(type.toInt()); bo.write(type.toInt());
ByteUtil.writeLong(bo, sendID); ByteUtil.writeLong(bo, sendID);
ByteUtil.writeLong(bo, order); ByteUtil.writeLong(bo, order);
try { try {
bo.write(content); bo.write(content);
} catch (IOException e) { } catch (IOException e) {
e.printStackTrace(); e.printStackTrace();
} }
byte[] ret = bo.toByteArray(); byte[] ret = bo.toByteArray();
int hashCode = ret.hashCode(); int hashCode = ret.hashCode();
return hashCode; return hashCode;
} }
public byte[] getBytes() { public byte[] getBytes() {
ByteArrayOutputStream bo = new ByteArrayOutputStream(); ByteArrayOutputStream bo = new ByteArrayOutputStream();
try { try {
bo.write(type.toInt()); bo.write(type.toInt());
ByteUtil.writeInt(bo, sendID); ByteUtil.writeInt(bo, sendID);
ByteUtil.writeLong(bo, order); ByteUtil.writeLong(bo, order);
if (signature != null) { if (signature != null) {
byte[] signature = this.signature.getBytes(); byte[] signature = this.signature.getBytes();
ByteUtil.writeInt(bo, signature.length); ByteUtil.writeInt(bo, signature.length);
bo.write(signature); bo.write(signature);
} else } else
ByteUtil.writeInt(bo, 0); ByteUtil.writeInt(bo, 0);
bo.write(content); bo.write(content);
} catch (IOException e) { } catch (IOException e) {
e.printStackTrace(); e.printStackTrace();
} }
return bo.toByteArray(); return bo.toByteArray();
} }
public static PBFTMessage parse(byte[] b) { public static PBFTMessage parse(byte[] b) {
PBFTMessage msg = new PBFTMessage(); PBFTMessage msg = new PBFTMessage();
ByteArrayInputStream bi = new ByteArrayInputStream(b); ByteArrayInputStream bi = new ByteArrayInputStream(b);
msg.type = PBFTType.fromByte(bi.read()); msg.type = PBFTType.fromByte(bi.read());
msg.sendID = ByteUtil.readInt(bi); msg.sendID = ByteUtil.readInt(bi);
msg.order = ByteUtil.readLong(bi); msg.order = ByteUtil.readLong(bi);
int sigLen = ByteUtil.readInt(bi); int sigLen = ByteUtil.readInt(bi);
if (sigLen > 0) if (sigLen > 0)
msg.signature = new String(ByteUtil.readBytes(bi, sigLen)); msg.signature = new String(ByteUtil.readBytes(bi, sigLen));
else else
msg.signature = null; msg.signature = null;
msg.content = ByteUtil.readBytes(bi, b.length - 1 - 4 - 8 - 4 - sigLen); msg.content = ByteUtil.readBytes(bi, b.length - 1 - 4 - 8 - 4 - sigLen);
return msg; return msg;
} }
private static String IDA = "PBFTMsg"; private static String IDA = "PBFTMsg";
// public void sign(SM2KeyPair pair) { // public void sign(SM2KeyPair pair) {
// sendID = Arrays.hashCode(pair.getPublicKey().getEncoded(false)); // sendID = Arrays.hashCode(pair.getPublicKey().getEncoded(false));
// signature = sm02.sign(content, IDA, pair).toString(); // signature = sm02.sign(content, IDA, pair).toString();
// } // }
// //
// public boolean verify(PBFTMember member) { // public boolean verify(PBFTMember member) {
// return sm02.verify(content, Signature.loadFromString(signature), IDA, // return sm02.verify(content, Signature.loadFromString(signature), IDA,
// SM2KeyPair.publicKeyStr2ECPoint(member.pubKey)); // SM2KeyPair.publicKeyStr2ECPoint(member.pubKey));
// } // }
public void setType(PBFTType type) { public void setType(PBFTType type) {
this.type = type; this.type = type;
} }
public void setContent(byte[] content) { public void setContent(byte[] content) {
this.content = content; this.content = content;
} }
public String getDisplayStr() { public String getDisplayStr() {
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
sb.append(sendID); sb.append(sendID);
sb.append("_"); sb.append("_");
sb.append(order); sb.append(order);
sb.append("_TYPE_").append(type); sb.append("_TYPE_").append(type);
sb.append(" --> sig:"); sb.append(" --> sig:");
sb.append(signature); sb.append(signature);
sb.append(" content:"); sb.append(" content:");
sb.append(new String(content)); sb.append(new String(content));
return sb.toString(); return sb.toString();
} }
public void setOrder(long order) { public void setOrder(long order) {
this.order = order; this.order = order;
} }
} }

57
src/main/java/org/bdware/sc/consistency/pbft/PBFTType.java
View File

@ -1,36 +1,35 @@
package org.bdware.sc.consistency.pbft; package org.bdware.sc.consistency.pbft;
public enum PBFTType { public enum PBFTType {
Request(0), PrePrepare(1), Prepare(2), Commit(3), Reply(4), Unknown(5), ReSend(6), AddMember(7); Request(0), PrePrepare(1), Prepare(2), Commit(3), Reply(4), Unknown(5), ReSend(6), AddMember(7);
private int type;
private int type; PBFTType(int i) {
type = i;
}
PBFTType(int i) { public static PBFTType fromByte(int i) {
type = i; switch (i) {
} case 0:
return Request;
case 1:
return PrePrepare;
case 2:
return Prepare;
case 3:
return Commit;
case 4:
return Reply;
case 6:
return ReSend;
case 7:
return AddMember;
default:
return Unknown;
}
}
public static PBFTType fromByte(int i) { public int toInt() {
switch (i) { return type;
case 0: }
return Request;
case 1:
return PrePrepare;
case 2:
return Prepare;
case 3:
return Commit;
case 4:
return Reply;
case 6:
return ReSend;
case 7:
return AddMember;
default:
return Unknown;
}
}
public int toInt() {
return type;
}
} }

7
src/main/java/org/bdware/sc/consistency/pbft/ViewAlgorithm.java
View File

@ -4,6 +4,8 @@ import org.bdware.sc.bean.ContractRequest;
import org.bdware.sc.conn.Node; import org.bdware.sc.conn.Node;
import org.bdware.sc.consistency.CommitAlgorithm; import org.bdware.sc.consistency.CommitAlgorithm;
import org.bdware.sc.consistency.Committer; import org.bdware.sc.consistency.Committer;
import org.bdware.sc.consistency.pbft.PBFTMessage;
import org.bdware.sc.consistency.pbft.PBFTType;
import org.bdware.sc.units.TrustfulExecutorConnection; import org.bdware.sc.units.TrustfulExecutorConnection;
public class ViewAlgorithm implements CommitAlgorithm { public class ViewAlgorithm implements CommitAlgorithm {
@ -11,7 +13,8 @@ public class ViewAlgorithm implements CommitAlgorithm {
private Committer committer; private Committer committer;
private TrustfulExecutorConnection connection; private TrustfulExecutorConnection connection;
public ViewAlgorithm(boolean isMaster) {} public ViewAlgorithm(boolean isMaster) {
}
public void setCommitter(Committer c) { public void setCommitter(Committer c) {
committer = c; committer = c;
@ -29,4 +32,4 @@ public class ViewAlgorithm implements CommitAlgorithm {
committer.onCommit(cr); committer.onCommit(cr);
} }
} }
} }

237
src/main/java/org/bdware/sc/event/EventBroker.java
View File

@ -34,7 +34,7 @@ public class EventBroker {
private final Map<String, IEventConsumer> id2Consumers; private final Map<String, IEventConsumer> id2Consumers;
private final Map<String, ThreadFlag> threadFlags; private final Map<String, ThreadFlag> threadFlags;
private final Map<String, Long> tempTopics; private final Map<String, Long> tempTopics;
// private final Map<String, Stack<REvent>> client2Events; // private final Map<String, Stack<REvent>> client2Events;
public EventBroker() { public EventBroker() {
center = new EventCenter(); center = new EventCenter();
@ -49,32 +49,43 @@ public class EventBroker {
tempTopics = recorder.recoverTempTopicsFromDb(); tempTopics = recorder.recoverTempTopicsFromDb();
// regularly check temporary topics and clean them // regularly check temporary topics and clean them
ContractManager.scheduledThreadPool.scheduleWithFixedDelay(() -> { ContractManager.scheduledThreadPool.scheduleWithFixedDelay(
long current = System.currentTimeMillis(); () -> {
int oldSize = tempTopics.size(); long current = System.currentTimeMillis();
tempTopics.keySet().forEach(topic -> { int oldSize = tempTopics.size();
if (tempTopics.get(topic) + EXPIRED_TIME > current) { tempTopics.keySet().forEach(topic -> {
String reqID = ContractManager.instance.nodeCenterConn.getNodeId() + "_" if (tempTopics.get(topic) + EXPIRED_TIME > current) {
+ System.currentTimeMillis(); String reqID =
REvent cleanEvent = new REvent(topic, UNSUBSCRIBE, null, reqID); ContractManager.instance.nodeCenterConn.getNodeId() +
cleanEvent "_" + System.currentTimeMillis();
.doSignature(ContractManager.instance.nodeCenterConn.getNodeKeyPair()); REvent cleanEvent =
handle(cleanEvent); new REvent(
tempTopics.remove(topic); topic,
} UNSUBSCRIBE,
}); null,
if (oldSize != tempTopics.size()) { reqID);
recorder.saveTempTopics(tempTopics); cleanEvent.doSignature(ContractManager.instance.nodeCenterConn.getNodeKeyPair());
} handle(cleanEvent);
}, 0L, EXPIRED_TIME, TimeUnit.MILLISECONDS); tempTopics.remove(topic);
}
});
if (oldSize != tempTopics.size()) {
recorder.saveTempTopics(tempTopics);
}
},
0L,
EXPIRED_TIME,
TimeUnit.MILLISECONDS);
// regularly create check point in database // regularly create check point in database
ContractManager.scheduledThreadPool.scheduleAtFixedRate( ContractManager.scheduledThreadPool.scheduleAtFixedRate(
() -> recorder.createCheckPoint(topic2cIds, id2Consumers), EXPIRED_TIME, () -> recorder.createCheckPoint(topic2cIds, id2Consumers),
EXPIRED_TIME, TimeUnit.MILLISECONDS); EXPIRED_TIME,
EXPIRED_TIME,
TimeUnit.MILLISECONDS);
NodeConsumer.setCenter(center); NodeConsumer.setCenter(center);
// client2Events = new HashMap<>(); // client2Events = new HashMap<>();
LOGGER.info("Event Broker starts!"); LOGGER.info("Event Broker starts!");
} }
@ -92,11 +103,10 @@ public class EventBroker {
switch (event.getType()) { switch (event.getType()) {
case SUBSCRIBE: case SUBSCRIBE:
if (null != topic && !topic.isEmpty()) { if (null != topic && !topic.isEmpty()) {
IEventConsumer consumer = doSubscribe(event); doSubscribe(event);
// save & try to sub in center // save & try to sub in center
recorder.appendEvent(event); recorder.appendEvent(event);
center.subInCenter(event.getTopic(), event.getSemantics(), event.getCenter(), center.subInCenter(event.getTopic(), event.getSemantics());
consumer, event);
} }
break; break;
case UNSUBSCRIBE: case UNSUBSCRIBE:
@ -106,10 +116,9 @@ public class EventBroker {
case PUBLISH: case PUBLISH:
case PREPUB: case PREPUB:
case PRESUB: case PRESUB:
LOGGER.info(String.format("Receive %s event from topic %s", event.getSemantics(), LOGGER.info(String.format("Receive %s event from topic %s", event.getSemantics(), topic));
topic)); LOGGER.debug(String.format("Receive %s event %s: %s",
LOGGER.debug(String.format("Receive %s event %s: %s", event.getSemantics(), topic, event.getSemantics(), topic, event.getContent()));
event.getContent()));
if (event.isForward()) { if (event.isForward()) {
// send event to the event center // send event to the event center
event.setForward(center.deliverEvent(topic, event)); event.setForward(center.deliverEvent(topic, event));
@ -124,17 +133,13 @@ public class EventBroker {
} }
} }
private IEventConsumer doSubscribe(REvent e) { private void doSubscribe(REvent e) {
IEventConsumer consumer = parseConsumer(e.getContent()); subInReg(e.getTopic(), parseConsumer(e.getContent()), this.topic2cIds, this.id2Consumers);
if (subInReg(e.getTopic(), consumer, this.topic2cIds, this.id2Consumers)) { // for events with semantics ONLY_ONCE, mark the topic is a temporary topic
// for events with semantics ONLY_ONCE, mark the topic is a temporary topic if (ONLY_ONCE.equals(e.getSemantics()) && !tempTopics.containsKey(e.getTopic())) {
if (ONLY_ONCE.equals(e.getSemantics()) && !tempTopics.containsKey(e.getTopic())) { tempTopics.put(e.getTopic(), System.currentTimeMillis());
tempTopics.put(e.getTopic(), System.currentTimeMillis()); recorder.saveTempTopics(tempTopics);
recorder.saveTempTopics(tempTopics);
}
return consumer;
} }
return null;
} }
public void doSubscribe(String topic, IEventConsumer consumer) { public void doSubscribe(String topic, IEventConsumer consumer) {
@ -144,14 +149,17 @@ public class EventBroker {
/** /**
* do subscribing in registry * do subscribing in registry
* *
* @param topic event topic * @param topic event topic
* @param consumer the consumer * @param consumer the consumer
* @param topic2cIds topic registry of broker or a check point in event recorder * @param topic2cIds topic registry of broker or a check point in event recorder
* @param id2Consumers consumer registry of broker or a check point in event recorder * @param id2Consumers consumer registry of broker or a check point in event recorder
* @return if the subscribing succeeds * @return if the subscribing succeeds
*/ */
public boolean subInReg(String topic, IEventConsumer consumer, public boolean subInReg(
Map<String, Set<String>> topic2cIds, Map<String, IEventConsumer> id2Consumers) { String topic,
IEventConsumer consumer,
Map<String, Set<String>> topic2cIds,
Map<String, IEventConsumer> id2Consumers) {
if (null == consumer) { if (null == consumer) {
return false; return false;
} }
@ -165,8 +173,8 @@ public class EventBroker {
topic2cIds.get(topic).add(cId); topic2cIds.get(topic).add(cId);
switch (consumer.getType()) { switch (consumer.getType()) {
case Contract: case Contract:
LOGGER.info("contract " + ((ContractConsumer) consumer).getContract() LOGGER.info("contract " + ((ContractConsumer) consumer).getContract() +
+ " subscribes topic " + topic); " subscribes topic " + topic);
break; break;
case Node: case Node:
LOGGER.info("node " + consumer.getId() + " subscribes topic " + topic); LOGGER.info("node " + consumer.getId() + " subscribes topic " + topic);
@ -187,22 +195,24 @@ public class EventBroker {
* do subscribing in registry<br/> * do subscribing in registry<br/>
* topic and consumer must not be null at the same time * topic and consumer must not be null at the same time
* <ul> * <ul>
* <li>if consumer is null and topic is not, it means the topic is a temporary topic, remove * <li>if consumer is null and topic is not, it means the topic is a temporary topic, remove it</li>
* it</li> * <li>if topic is null and consumer is not,
* <li>if topic is null and consumer is not, it means a consumer or a contract wants to * it means a consumer or a contract wants to unsubscribe all topics,
* unsubscribe all topics, remove all related consumers in topic registry and consumer * remove all related consumers in topic registry and consumer registry</li>
* registry</li> * <li>if two of them is not null, do unsubscribing in two registries</li>
* <li>if two of them is not null, do unsubscribing in two registries</li>
* </ul> * </ul>
* *
* @param topic event topic * @param topic event topic
* @param consumer the consumer, just id is required * @param consumer the consumer, just id is required
* @param topic2cIds topic registry of broker or a check point in event recorder * @param topic2cIds topic registry of broker or a check point in event recorder
* @param id2Consumers consumer registry of broker or a check point in event recorder * @param id2Consumers consumer registry of broker or a check point in event recorder
* @return if the subscribing succeeds * @return if the subscribing succeeds
*/ */
public boolean unsubInReg(String topic, IEventConsumer consumer, public boolean unsubInReg(
Map<String, Set<String>> topic2cIds, Map<String, IEventConsumer> id2Consumers) { String topic,
IEventConsumer consumer,
Map<String, Set<String>> topic2cIds,
Map<String, IEventConsumer> id2Consumers) {
if (null == topic && null == consumer) { if (null == topic && null == consumer) {
return false; return false;
} }
@ -222,8 +232,7 @@ public class EventBroker {
// if cId belongs to a contract, find all related consumers // if cId belongs to a contract, find all related consumers
toRmIds = new ArrayList<>(); toRmIds = new ArrayList<>();
id2Consumers.forEach((k, c) -> { id2Consumers.forEach((k, c) -> {
if (c instanceof ContractConsumer if (c instanceof ContractConsumer && ((ContractConsumer) c).getContract().equals(cId)) {
&& ((ContractConsumer) c).getContract().equals(cId)) {
toRmIds.add(k); toRmIds.add(k);
} }
}); });
@ -246,8 +255,7 @@ public class EventBroker {
/** /**
* parse consumer information from content str<br/> * parse consumer information from content str<br/>
* if caller wants to select all consumers of a contract, the content str is also parsed into a * if caller wants to select all consumers of a contract, the content str is also parsed into a node consumer
* node consumer
* *
* @param content json string, {"subscriber": "[subscriber]", "handler?": "[handler]"} * @param content json string, {"subscriber": "[subscriber]", "handler?": "[handler]"}
* @return a node consumer or contract consumer, or null if exception is thrown * @return a node consumer or contract consumer, or null if exception is thrown
@ -267,9 +275,6 @@ public class EventBroker {
if (null != handler && !handler.isEmpty()) { if (null != handler && !handler.isEmpty()) {
consumer = new ContractConsumer(subscriber, handler); consumer = new ContractConsumer(subscriber, handler);
} else { } else {
if (subscriber.equals(ContractManager.instance.nodeCenterConn.getNodeId())) {
return null;
}
consumer = new NodeConsumer(subscriber); consumer = new NodeConsumer(subscriber);
} }
return consumer; return consumer;
@ -293,16 +298,19 @@ public class EventBroker {
case AT_LEAST_ONCE: case AT_LEAST_ONCE:
case NEED_RETRY: case NEED_RETRY:
// send events to all // send events to all
topicConsumers topicConsumers.forEach(cId ->
.forEach(cId -> deliverEvent(event, cEvent, nEventStr, cId, topic, true)); deliverEvent(event, cEvent, nEventStr, cId, topic, true));
break; break;
case AT_MOST_ONCE: case AT_MOST_ONCE:
// send event to a random consumer // send event to a random consumer
// AT_MOST_ONCE, so broker don't need to do anything when delivering fails // AT_MOST_ONCE, so broker don't need to do anything when delivering fails
deliverEvent(event, cEvent, nEventStr, deliverEvent(
topicConsumers.toArray()[(int) (Math.random() * topicConsumers.size())] event,
.toString(), cEvent,
topic, true); nEventStr,
topicConsumers.toArray()[(int) (Math.random() * topicConsumers.size())].toString(),
topic,
true);
break; break;
case ONLY_ONCE: case ONLY_ONCE:
switch (event.getType()) { switch (event.getType()) {
@ -328,14 +336,20 @@ public class EventBroker {
// send PREPUB events to all consumers // send PREPUB events to all consumers
// TODO if there are no consumers to receive the ONLY_ONCE events? // TODO if there are no consumers to receive the ONLY_ONCE events?
ContractManager.threadPool.execute(() -> { ContractManager.threadPool.execute(() -> {
REvent prePubMsg = new REvent(event.getTopic(), PREPUB, contentHash, REvent prePubMsg = new REvent(event.getTopic(),
PREPUB,
contentHash,
event.getRequestID()); event.getRequestID());
prePubMsg.doSignature( prePubMsg.doSignature(ContractManager.instance.nodeCenterConn.getNodeKeyPair());
ContractManager.instance.nodeCenterConn.getNodeKeyPair()); topicConsumers.forEach(cId ->
topicConsumers.forEach(cId -> deliverEvent(prePubMsg, deliverEvent(prePubMsg,
new Event(event.getTopic(), contentHash, new Event(event.getTopic(),
prePubMsg.getSemantics()), contentHash,
JsonUtil.toJson(prePubMsg), cId, topic, false)); prePubMsg.getSemantics()),
JsonUtil.toJson(prePubMsg),
cId,
topic,
false));
// wait for responses from contracts (PRESUB events) // wait for responses from contracts (PRESUB events)
while (true) { while (true) {
try { try {
@ -343,21 +357,21 @@ public class EventBroker {
flag.wait(30 * 1000L); flag.wait(30 * 1000L);
} }
if (!flag.get().isEmpty()) { if (!flag.get().isEmpty()) {
REvent finalMsg = new REvent(flag.get(), PUBLISH, REvent finalMsg = new REvent(flag.get(),
event.getContent(), HashUtil.sha3(contentHash PUBLISH,
+ System.currentTimeMillis())); event.getContent(),
HashUtil.sha3(
contentHash + System.currentTimeMillis()));
// if the delivering fails, retry publishing // if the delivering fails, retry publishing
finalMsg.setSemantics(NEED_RETRY); finalMsg.setSemantics(NEED_RETRY);
finalMsg.doSignature( finalMsg.doSignature(
ContractManager.instance.nodeCenterConn ContractManager.instance.nodeCenterConn.getNodeKeyPair());
.getNodeKeyPair());
handle(finalMsg); handle(finalMsg);
break; break;
} }
} catch (InterruptedException e) { } catch (InterruptedException e) {
LOGGER.warn("ONLY_ONE event delivering is interrupted: " LOGGER.warn("ONLY_ONE event delivering is interrupted: " + e.getMessage());
+ e.getMessage()); // e.printStackTrace();
// e.printStackTrace();
} }
} }
}); });
@ -374,16 +388,20 @@ public class EventBroker {
/** /**
* publish the event to the consumer * publish the event to the consumer
* *
* @param event event message * @param event event message
* @param cEvent simple event message to the contract consumer, only topic, content and * @param cEvent simple event message to the contract consumer, only topic, content and semantics
* semantics
* @param nEventStr event message to the node * @param nEventStr event message to the node
* @param cId consumer id * @param cId consumer id
* @param topic topic of the event * @param topic topic of the event
* @param isPub if the event is published or pre-published * @param isPub if the event is published or pre-published
*/ */
private void deliverEvent(REvent event, Event cEvent, String nEventStr, String cId, private void deliverEvent(
String topic, boolean isPub) { REvent event,
Event cEvent,
String nEventStr,
String cId,
String topic,
boolean isPub) {
if (id2Consumers.containsKey(cId)) { if (id2Consumers.containsKey(cId)) {
IEventConsumer consumer = id2Consumers.get(cId); IEventConsumer consumer = id2Consumers.get(cId);
if (consumer instanceof ContractConsumer) { if (consumer instanceof ContractConsumer) {
@ -394,24 +412,31 @@ public class EventBroker {
public void onResult(String unused) { public void onResult(String unused) {
// if the delivering fails, unsubscribe the consumer // if the delivering fails, unsubscribe the consumer
ContractConsumer c = (ContractConsumer) consumer; ContractConsumer c = (ContractConsumer) consumer;
ContractClient client = ContractClient client = ContractManager.instance.getClient(c.getContract());
ContractManager.instance.getClient(c.getContract()); String reqID =
String reqID = ContractManager.instance.nodeCenterConn.getNodeKeyPair() ContractManager.instance.nodeCenterConn.getNodeKeyPair().getPublicKeyStr() +
.getPublicKeyStr() + "_" + System.currentTimeMillis(); "_" + System.currentTimeMillis();
REvent unsubEvent = new REvent(topic, UNSUBSCRIBE, REvent unsubEvent =
"{\"subscriber\":\"" + cId + "\"}", reqID); new REvent(
topic,
UNSUBSCRIBE,
"{\"subscriber\":\"" + cId + "\"}",
reqID);
unsubEvent.doSignature(client.getPubkey(), client.getContractKey()); unsubEvent.doSignature(client.getPubkey(), client.getContractKey());
handle(unsubEvent); handle(unsubEvent);
// if the event is an ONLY_ONCE event, retry publishing // if the event is an ONLY_ONCE event, retry publishing
if (NEED_RETRY.equals(event.getSemantics())) { if (NEED_RETRY.equals(event.getSemantics())) {
REvent newMsg = new REvent(topic.split("\\|")[0], PUBLISH, REvent newMsg =
event.getContent(), event.getRequestID()); new REvent(
topic.split("\\|")[0],
PUBLISH,
event.getContent(),
event.getRequestID());
newMsg.setSemantics(ONLY_ONCE); newMsg.setSemantics(ONLY_ONCE);
newMsg.setHash(event.getHash()); newMsg.setHash(event.getHash());
newMsg.setTxHash(event.getTxHash()); newMsg.setTxHash(event.getTxHash());
newMsg.doSignature( newMsg.doSignature(ContractManager.instance.nodeCenterConn.getNodeKeyPair());
ContractManager.instance.nodeCenterConn.getNodeKeyPair());
handle(newMsg); handle(newMsg);
} }
} }
@ -433,8 +458,8 @@ public class EventBroker {
}); });
} else if (isPub) { } else if (isPub) {
// client consumer // client consumer
ContractManager.threadPool ContractManager.threadPool.execute(() ->
.execute(() -> consumer.publishEvent(nEventStr, new ResultCallback() { consumer.publishEvent(nEventStr, new ResultCallback() {
@Override @Override
public void onResult(String unused) { public void onResult(String unused) {
unsubInReg(null, consumer, topic2cIds, id2Consumers); unsubInReg(null, consumer, topic2cIds, id2Consumers);
@ -469,4 +494,4 @@ public class EventBroker {
this.flag = flag; this.flag = flag;
} }
} }
} }

61
src/main/java/org/bdware/sc/event/EventCenter.java
View File

@ -1,8 +1,5 @@
package org.bdware.sc.event; package org.bdware.sc.event;
import org.bdware.sc.event.clients.ContractConsumer;
import org.bdware.sc.event.clients.IEventConsumer;
import org.bdware.sc.event.clients.NodeConsumer;
import org.bdware.sc.util.DHTUtil; import org.bdware.sc.util.DHTUtil;
import org.bdware.sc.util.JsonUtil; import org.bdware.sc.util.JsonUtil;
@ -12,64 +9,45 @@ import static org.bdware.sc.ContractManager.instance;
* @author Kaidong Wu * @author Kaidong Wu
*/ */
public class EventCenter { public class EventCenter {
// private static final Logger LOGGER = LogManager.getLogger(EventCenter.class); // private static final Logger LOGGER = LogManager.getLogger(EventCenter.class);
/** /**
* get the nearest node to the topic in the hash function range * get the nearest node to the topic in the hash function range
* *
* @param topic the topic * @param topic the topic
* @param k the number of centers
* @return id of the node * @return id of the node
*/ */
public String[] getCenterByTopic(String topic, int k) { public String getCenterByTopic(String topic) {
if (null == instance.nodeCenterConn) { if (null == instance.nodeCenterConn) {
return null; return null;
} }
String[] centers = DHTUtil.getClusterByKey(topic, k); String[] centers = DHTUtil.getClusterByKey(topic, 1);
if (null == centers || centers.length == 0) { if (null == centers || centers.length == 0) {
return null; return null;
} }
return centers; return centers[0];
} }
/** /**
* subscribe a topic in center * subscribe a topic in center
* *
* @param topic event topic * @param topic event topic
* @param semantics event semantics, used to mark PRESUB events * @param semantics event semantics, used to mark PRESUB events
* @param center id of event center if the subscribing has been handled
* @param consumer consumer
* @param event original event
*/ */
public void subInCenter(String topic, REvent.REventSemantics semantics, String center, public void subInCenter(String topic, REvent.REventSemantics semantics) {
IEventConsumer consumer, REvent event) {
if (null == instance.nodeCenterConn) { if (null == instance.nodeCenterConn) {
return; return;
} }
REvent msg = new REvent(topic, REvent.REventType.SUBSCRIBE, REvent msg = new REvent(
String.format("{\"subscriber\":\"%s\"}", instance.nodeCenterConn.getNodeId()), ""); topic,
REvent.REventType.SUBSCRIBE,
String.format("{\"subscriber\":\"%s\"}",
instance.nodeCenterConn.getNodeId()),
"");
msg.setSemantics(semantics); msg.setSemantics(semantics);
msg.doSignature(instance.nodeCenterConn.getNodeKeyPair()); msg.doSignature(instance.nodeCenterConn.getNodeKeyPair());
msg.setCenter(center); String nodeId = getCenterByTopic(topic);
String[] centers = getCenterByTopic(topic, 2); instance.masterStub.deliverEvent(nodeId, JsonUtil.toJson(msg));
if (null == centers) {
return;
}
if (null != center) {
if (centers[0].equals(center) && centers.length > 1) {
instance.masterStub.deliverEvent(centers[1], JsonUtil.toJson(msg));
} else {
instance.masterStub.deliverEvent(centers[0], JsonUtil.toJson(msg));
}
} else if (!instance.masterStub.deliverEvent(centers[0], JsonUtil.toJson(msg))) {
if (consumer instanceof NodeConsumer) {
event.setCenter(centers[0]);
instance.masterStub.deliverEvent(consumer.getId(), JsonUtil.toJson(event));
} else if (consumer instanceof ContractConsumer && centers.length > 1) {
msg.setCenter(centers[0]);
instance.masterStub.deliverEvent(centers[1], JsonUtil.toJson(msg));
}
}
} }
/** /**
@ -83,22 +61,19 @@ public class EventCenter {
if (null == instance.masterStub) { if (null == instance.masterStub) {
return false; return false;
} }
String[] centers = getCenterByTopic(topic, 1); String nodeId = getCenterByTopic(topic);
if (null == centers) { return instance.masterStub.deliverEvent(nodeId, JsonUtil.toJson(event));
return false;
}
return instance.masterStub.deliverEvent(centers[0], JsonUtil.toJson(event));
} }
/** /**
* publish an event to another node; used by NodeConsumer * publish an event to another node; used by NodeConsumer
* *
* @param target id of the target node * @param target id of the target node
* @param eStr event string * @param eStr event string
*/ */
public void publishEvent(String target, String eStr) { public void publishEvent(String target, String eStr) {
if (null != instance.masterStub) { if (null != instance.masterStub) {
instance.masterStub.deliverEvent(target, eStr); instance.masterStub.deliverEvent(target, eStr);
} }
} }
} }

33
src/main/java/org/bdware/sc/event/EventRecorder.java
View File

@ -53,7 +53,7 @@ public class EventRecorder {
} }
public void createCheckPoint(Map<String, Set<String>> topic2cIds, public void createCheckPoint(Map<String, Set<String>> topic2cIds,
Map<String, IEventConsumer> id2Consumers) { Map<String, IEventConsumer> id2Consumers) {
CheckPoint cp = new CheckPoint(topic2cIds, id2Consumers); CheckPoint cp = new CheckPoint(topic2cIds, id2Consumers);
synchronized (latestEvent) { synchronized (latestEvent) {
if (!latestEvent.isCp()) { if (!latestEvent.isCp()) {
@ -79,11 +79,9 @@ public class EventRecorder {
String key = KeyValueRocksDBUtil.instance.getValue(dbName, LATEST_EVENT_KEY); String key = KeyValueRocksDBUtil.instance.getValue(dbName, LATEST_EVENT_KEY);
latestEvent.set(key); latestEvent.set(key);
CheckPoint cp = new CheckPoint(); CheckPoint cp = new CheckPoint();
Type topic2cIdsType = Type topic2cIdsType = TypeToken.getParameterized(ConcurrentHashMap.class, String.class,
TypeToken TypeToken.getParameterized(Set.class, String.class,
.getParameterized(ConcurrentHashMap.class, String.class, String.class).getType()).getType();
TypeToken.getParameterized(Set.class, String.class).getType())
.getType();
// retrieving transactions from database // retrieving transactions from database
while (null != key && !key.isEmpty()) { while (null != key && !key.isEmpty()) {
String json = KeyValueRocksDBUtil.instance.getValue(dbName, key); String json = KeyValueRocksDBUtil.instance.getValue(dbName, key);
@ -95,16 +93,14 @@ public class EventRecorder {
if (json.startsWith("cp")) { if (json.startsWith("cp")) {
// create check point by the transaction and stop retrieving // create check point by the transaction and stop retrieving
JsonObject data = JsonUtil.parseStringAsJsonObject(json.substring(2)); JsonObject data = JsonUtil.parseStringAsJsonObject(json.substring(2));
cp.topic2cIds = cp.topic2cIds = JsonUtil.fromJson(data.get("topic2cIds").toString(), topic2cIdsType);
JsonUtil.fromJson(data.get("topic2cIds").toString(), topic2cIdsType);
JsonObject id2Consumers = data.getAsJsonObject("id2Consumers"); JsonObject id2Consumers = data.getAsJsonObject("id2Consumers");
for (String k : id2Consumers.keySet()) { for (String k : id2Consumers.keySet()) {
JsonObject consumer = id2Consumers.getAsJsonObject(k); JsonObject consumer = id2Consumers.getAsJsonObject(k);
if (!consumer.has("type")) { if (!consumer.has("type")) {
continue; continue;
} }
ConsumerType type = ConsumerType type = ConsumerType.valueOf(consumer.get("type").getAsString());
ConsumerType.valueOf(consumer.get("type").getAsString());
switch (type) { switch (type) {
case Contract: case Contract:
cp.id2Consumers.put(k, cp.id2Consumers.put(k,
@ -149,8 +145,7 @@ public class EventRecorder {
// if empty, return the check point // if empty, return the check point
latestEvent.setCp(true); latestEvent.setCp(true);
} else { } else {
// on the base of old check point, process following sub or unsub events to recover // on the base of old check point, process following sub or unsub events to recover registry
// registry
while (!stack.empty()) { while (!stack.empty()) {
Object record = stack.pop(); Object record = stack.pop();
if (record instanceof CheckPoint) { if (record instanceof CheckPoint) {
@ -160,19 +155,15 @@ public class EventRecorder {
IEventConsumer consumer = broker.parseConsumer(tran.content); IEventConsumer consumer = broker.parseConsumer(tran.content);
switch (tran.type) { switch (tran.type) {
case SUBSCRIBE: case SUBSCRIBE:
if (!broker.subInReg(tran.topic, consumer, cp.topic2cIds, if (!broker.subInReg(tran.topic, consumer, cp.topic2cIds, cp.id2Consumers)) {
cp.id2Consumers)) {
LOGGER.warn("record damaged! " + key); LOGGER.warn("record damaged! " + key);
LOGGER.debug( LOGGER.debug("record damaged! " + key + ": " + JsonUtil.toJson(tran));
"record damaged! " + key + ": " + JsonUtil.toJson(tran));
} }
break; break;
case UNSUBSCRIBE: case UNSUBSCRIBE:
if (!broker.unsubInReg(tran.topic, consumer, cp.topic2cIds, if (!broker.unsubInReg(tran.topic, consumer, cp.topic2cIds, cp.id2Consumers)) {
cp.id2Consumers)) {
LOGGER.warn("record damaged! " + key); LOGGER.warn("record damaged! " + key);
LOGGER.debug( LOGGER.debug("record damaged! " + key + ": " + JsonUtil.toJson(tran));
"record damaged! " + key + ": " + JsonUtil.toJson(tran));
} }
default: default:
break; break;
@ -234,7 +225,7 @@ public class EventRecorder {
} }
public CheckPoint(Map<String, Set<String>> topic2cIds, public CheckPoint(Map<String, Set<String>> topic2cIds,
Map<String, IEventConsumer> id2Consumers) { Map<String, IEventConsumer> id2Consumers) {
this.topic2cIds = topic2cIds; this.topic2cIds = topic2cIds;
this.id2Consumers = id2Consumers; this.id2Consumers = id2Consumers;
} }

22
src/main/java/org/bdware/sc/event/clients/ContractConsumer.java
View File

@ -1,6 +1,5 @@
package org.bdware.sc.event.clients; package org.bdware.sc.event.clients;
import com.google.gson.Gson;
import com.google.gson.annotations.Expose; import com.google.gson.annotations.Expose;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
@ -81,7 +80,6 @@ public class ContractConsumer implements IEventConsumer {
return; return;
} }
AtomicInteger callCount = new AtomicInteger(0); AtomicInteger callCount = new AtomicInteger(0);
LOGGER.info("ContractConsumer!" + new Gson().toJson(cr));
// TODO sending requests at a high frequency maybe cause that some requests are ignored // TODO sending requests at a high frequency maybe cause that some requests are ignored
ScheduledFuture<?> future = ContractManager.scheduledThreadPool.scheduleAtFixedRate( ScheduledFuture<?> future = ContractManager.scheduledThreadPool.scheduleAtFixedRate(
() -> ContractManager.instance.executeContractInternal(cr, new ResultCallback() { () -> ContractManager.instance.executeContractInternal(cr, new ResultCallback() {
@ -91,20 +89,19 @@ public class ContractConsumer implements IEventConsumer {
try { try {
ContractResult result = JsonUtil.fromJson(str, ContractResult.class); ContractResult result = JsonUtil.fromJson(str, ContractResult.class);
if (!result.status.equals(ContractResult.Status.Success)) { if (!result.status.equals(ContractResult.Status.Success)) {
if (callCount.get() == TIMEOUT_COUNT if (callCount.get() == TIMEOUT_COUNT ||
|| (result.status == ContractResult.Status.Exception (result.status == ContractResult.Status.Exception &&
&& result.result.toString() result.result.toString().contains("not exported"))) {
.contains("not exported"))) { LOGGER.error(String.format(
LOGGER.error(String.format("receiving event error! %s.%s: %s", "receiving event error! %s.%s: %s", contract, handler, str));
contract, handler, str));
rc.onResult((String) null); rc.onResult((String) null);
} else { } else {
ret = false; ret = false;
} }
} }
} catch (Exception e) { } catch (Exception e) {
LOGGER.error(String.format("receiving event error! %s.%s: %s", contract, LOGGER.error(String.format(
handler, e.getMessage())); "receiving event error! %s.%s: %s", contract, handler, e.getMessage()));
rc.onResult((String) null); rc.onResult((String) null);
} }
callCount.incrementAndGet(); callCount.incrementAndGet();
@ -122,7 +119,10 @@ public class ContractConsumer implements IEventConsumer {
} }
} }
}, (reqID, hashStr) -> { }, (reqID, hashStr) -> {
}), 500L, 2500L, TimeUnit.MILLISECONDS); }),
500L,
2500L,
TimeUnit.MILLISECONDS);
scheduledFutures.put(getId(), future); scheduledFutures.put(getId(), future);
synchronized (flag) { synchronized (flag) {
flag.notify(); flag.notify();

5
src/main/java/org/bdware/sc/event/clients/IEventConsumer.java
View File

@ -1,6 +1,7 @@
package org.bdware.sc.event.clients; package org.bdware.sc.event.clients;
import org.bdware.sc.conn.ResultCallback; import org.bdware.sc.conn.ResultCallback;
import org.bdware.sc.event.Event;
public interface IEventConsumer { public interface IEventConsumer {
String getId(); String getId();
@ -12,6 +13,8 @@ public interface IEventConsumer {
void competeSub(Object msg, ResultCallback rc, String... options); void competeSub(Object msg, ResultCallback rc, String... options);
enum ConsumerType { enum ConsumerType {
Contract, Node, WSClient Contract,
Node,
WSClient
} }
} }

2
src/main/java/org/bdware/sc/event/clients/NodeConsumer.java
View File

@ -43,4 +43,4 @@ public class NodeConsumer implements IEventConsumer {
public void competeSub(Object msg, ResultCallback rc, String... options) { public void competeSub(Object msg, ResultCallback rc, String... options) {
publishEvent(msg, rc, options); publishEvent(msg, rc, options);
} }
} }

3
src/main/java/org/bdware/sc/event/clients/WSClientConsumer.java
View File

@ -47,5 +47,6 @@ public class WSClientConsumer implements IEventConsumer {
} }
@Override @Override
public void competeSub(Object msg, ResultCallback rc, String... options) {} public void competeSub(Object msg, ResultCallback rc, String... options) {
}
} }

10
src/main/java/org/bdware/sc/handler/ManagerHandler.java
View File

@ -6,6 +6,7 @@ import org.bdware.sc.bean.ContractRequest;
import org.bdware.sc.conn.Description; import org.bdware.sc.conn.Description;
import org.bdware.sc.conn.MsgHandler; import org.bdware.sc.conn.MsgHandler;
import org.bdware.sc.conn.ResultCallback; import org.bdware.sc.conn.ResultCallback;
import org.bdware.sc.event.REvent;
import org.bdware.sc.get.GetMessage; import org.bdware.sc.get.GetMessage;
import org.bdware.sc.util.JsonUtil; import org.bdware.sc.util.JsonUtil;
@ -34,7 +35,8 @@ public class ManagerHandler extends MsgHandler {
cb.onResult(cm.getTimesOfExecution(msg.arg)); cb.onResult(cm.getTimesOfExecution(msg.arg));
} }
@Description(value = "execute Contract, {\"contractID\":\"112233\",\"arg\":\"\"}", @Description(
value = "execute Contract, {\"contractID\":\"112233\",\"arg\":\"\"}",
isAsync = true) isAsync = true)
public void executeContract(GetMessage msg, ResultCallback cb) { public void executeContract(GetMessage msg, ResultCallback cb) {
ContractRequest c; ContractRequest c;
@ -87,10 +89,4 @@ public class ManagerHandler extends MsgHandler {
ContractRequest cr = JsonUtil.fromJson(msg.arg, ContractRequest.class); ContractRequest cr = JsonUtil.fromJson(msg.arg, ContractRequest.class);
cb.onResult(cm.addDEMember(cr.getContractID(), cr.getArg().getAsString())); cb.onResult(cm.addDEMember(cr.getContractID(), cr.getArg().getAsString()));
} }
@Description("add distributed execution member, sample: {contractID:id/name, arg:ipAndPort}")
public void getLedgerParams(GetMessage msg, ResultCallback cb) {
cb.onResult(ContractManager.instance.getLedgerParams().toString());
}
} }

120
src/main/java/org/bdware/sc/units/ContractRecord.java
View File

@ -4,51 +4,81 @@ package org.bdware.sc.units;
import java.io.*; import java.io.*;
/* /*
* 节点启动后记录自己的units信息和合约的一些信息 节点重新上线后，NC发要求让其恢复，则从本地读取自己的这些信息进行恢复 * 节点启动后记录自己的units信息和合约的一些信息
* 节点重新上线后，NC发要求让其恢复，则从本地读取自己的这些信息进行恢复
*/ */
public class ContractRecord implements Serializable { public class ContractRecord implements Serializable {
/* /* private static final long serialVersionUID = 704775241674568688L;
* private static final long serialVersionUID = 704775241674568688L;
* public transient RecoverFlag recoverFlag = RecoverFlag.Fine;
* public transient RecoverFlag recoverFlag = RecoverFlag.Fine;
* public String contractID;
* public String contractID; public String contractName; public String key; public String public String contractName;
* contractpubKey; public ContractType type; public int copies = 1; public int lastExeSeq = -1; public String key;
* public boolean isPrivate = false; public String pubKeyPath; public String ypkName; public String contractpubKey;
* public ContractType type;
* //public Map<String,String> members; //k-udpid,v-udpaddress public int copies = 1;
* public int lastExeSeq = -1;
* public String memory = ""; public boolean isPrivate = false;
* public String pubKeyPath;
* //JavaScriptEntry public String invokeID; public String ypkName;
*
* public ContractRecord(String id){ this.contractID = id; } //public Map<String,String> members; //k-udpid,v-udpaddress
*
* public ContractRecord(String id,String contractName,String key,String public String memory = "";
* contractpubKey,ContractType type,int copies){ this.contractID = id; this.contractName =
* contractName; this.key = key; this.contractpubKey = contractpubKey; this.type = type; //JavaScriptEntry
* this.copies = copies; } public String invokeID;
*
* public void printContent(){ System.out.println("==========ContractRecord========"); public ContractRecord(String id){
* this.contractID = id;
* System.out.println("contractID=" + contractID == null ? "null" : contractID); }
* System.out.println("contractName=" + contractName == null ? "null" : contractName);
* System.out.println("key=" + key == null ? "null" : key); System.out.println("contractPubKey=" public ContractRecord(String id,String contractName,String key,String contractpubKey,ContractType type,int copies){
* + contractpubKey == null ? "null" : contractpubKey); System.out.println("type=" + type == this.contractID = id;
* null ? "null" : type); System.out.println("lastExeSeq=" + lastExeSeq == null ? "null" : this.contractName = contractName;
* lastExeSeq); System.out.println("invokeID=" + invokeID == null ? "null" : invokeID); this.key = key;
* System.out.println("copies=" + copies); System.out.println("isPrivate=" + isPrivate); this.contractpubKey = contractpubKey;
* System.out.println("pubKeyPath=" + pubKeyPath == null ? "null" : pubKeyPath); this.type = type;
* System.out.println("ypkName=" + ypkName == null ? "null" : ypkName); if(members != null){ this.copies = copies;
* for(String k : members.keySet()){ System.out.println("members " + k + "-" + members.get(k)); }
* } } System.out.println("memory=" + memory == null ? "null" : memory);
* System.out.println("==========ContractRecord print finish======="); } public void printContent(){
* System.out.println("==========ContractRecord========");
* public static void getContentFromFile(String path){ File file = new File(path); try{
* FileInputStream os = new FileInputStream(file); ObjectInputStream oos = new System.out.println("contractID=" + contractID == null ? "null" : contractID);
* ObjectInputStream(os); ContractRecord record = (ContractRecord) oos.readObject(); System.out.println("contractName=" + contractName == null ? "null" : contractName);
* record.printContent(); } catch (FileNotFoundException e) { e.printStackTrace(); } catch System.out.println("key=" + key == null ? "null" : key);
* (IOException e) { e.printStackTrace(); } catch (ClassNotFoundException e) { System.out.println("contractPubKey=" + contractpubKey == null ? "null" : contractpubKey);
* e.printStackTrace(); } } System.out.println("type=" + type == null ? "null" : type);
*/ System.out.println("lastExeSeq=" + lastExeSeq == null ? "null" : lastExeSeq);
System.out.println("invokeID=" + invokeID == null ? "null" : invokeID);
System.out.println("copies=" + copies);
System.out.println("isPrivate=" + isPrivate);
System.out.println("pubKeyPath=" + pubKeyPath == null ? "null" : pubKeyPath);
System.out.println("ypkName=" + ypkName == null ? "null" : ypkName);
if(members != null){
for(String k : members.keySet()){
System.out.println("members " + k + "-" + members.get(k));
}
}
System.out.println("memory=" + memory == null ? "null" : memory);
System.out.println("==========ContractRecord print finish=======");
}
public static void getContentFromFile(String path){
File file = new File(path);
try{
FileInputStream os = new FileInputStream(file);
ObjectInputStream oos = new ObjectInputStream(os);
ContractRecord record = (ContractRecord) oos.readObject();
record.printContent();
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
}*/
} }

42
src/main/java/org/bdware/sc/units/ContractUnitController.java
View File

@ -12,11 +12,13 @@ import java.util.*;
public class ContractUnitController { public class ContractUnitController {
private static final Logger LOGGER = LogManager.getLogger(ContractUnitController.class); private static final Logger LOGGER = LogManager.getLogger(ContractUnitController.class);
private final TrustfulExecutorConnection connection; private final TrustfulExecutorConnection connection;
private final Map<String, ContractUnit> units; // 这个节点上<合约id，合约集群>，其中每个ContractUnit中存储了对应合约id的其他节点udp信息 private final Map<String, ContractUnit> units; //这个节点上<合约id，合约集群>，其中每个ContractUnit中存储了对应合约id的其他节点udp信息
ContractManager manager; ContractManager manager;
SequencingAlgorithmFactory algorithmFactory; SequencingAlgorithmFactory algorithmFactory;
public ContractUnitController(TrustfulExecutorConnection connection, ContractManager manager, public ContractUnitController(
TrustfulExecutorConnection connection,
ContractManager manager,
SequencingAlgorithmFactory factory) { SequencingAlgorithmFactory factory) {
this.connection = connection; this.connection = connection;
this.manager = manager; this.manager = manager;
@ -37,13 +39,18 @@ public class ContractUnitController {
unit.node2member = new HashMap<>(); unit.node2member = new HashMap<>();
units.put(req.contract.getID(), unit); units.put(req.contract.getID(), unit);
System.out.println("[ContractUnitController] startContract:" + result + " isMaster:" System.out.println(
+ req.isMaster + " cid:" + req.contract.getID()); "[ContractUnitController] startContract:"
+ result
+ " isMaster:"
+ req.isMaster
+ " cid:"
+ req.contract.getID());
break; break;
case AddMember: case AddMember:
LOGGER.debug("contractID:" + cumsg.getContractID()); LOGGER.debug("contractID:" + cumsg.getContractID());
unit = units.get(cumsg.getContractID()); unit = units.get(cumsg.getContractID());
unit.addMember(node, cumsg); // 启动通过在UDPTrustExecutor中遍历memebers，使得这个合约集群中每个节点都有其他节点的UDP信息，便于之后护发UDP消息 unit.addMember(node, cumsg); //启动通过在UDPTrustExecutor中遍历memebers，使得这个合约集群中每个节点都有其他节点的UDP信息，便于之后护发UDP消息
break; break;
case Sequencing: case Sequencing:
unit = units.get(cumsg.getContractID()); unit = units.get(cumsg.getContractID());
@ -52,7 +59,7 @@ public class ContractUnitController {
break; break;
case Unknown: case Unknown:
default: default:
// recover start //recover start
ContractUnitStartRequest req2 = ContractUnitStartRequest.parse(cumsg.content); ContractUnitStartRequest req2 = ContractUnitStartRequest.parse(cumsg.content);
unit = new ContractUnit(); unit = new ContractUnit();
unit.connection = connection; unit.connection = connection;
@ -61,8 +68,12 @@ public class ContractUnitController {
unit.node2member = new HashMap<>(); unit.node2member = new HashMap<>();
units.put(req2.contract.getID(), unit); units.put(req2.contract.getID(), unit);
System.out.println("[ContractUnitController] startContract:" + " isMaster:" System.out.println(
+ req2.isMaster + " cid:" + req2.contract.getID()); "[ContractUnitController] startContract:"
+ " isMaster:"
+ req2.isMaster
+ " cid:"
+ req2.contract.getID());
break; break;
} }
} }
@ -73,7 +84,7 @@ public class ContractUnitController {
public static class ContractUnit implements TrustfulExecutorConnection, Serializable { public static class ContractUnit implements TrustfulExecutorConnection, Serializable {
public String contractID; public String contractID;
public Map<Node, ContractUnitMember> node2member; // 存储和自己在一个合约集群的其他节点信息 public Map<Node, ContractUnitMember> node2member; //存储和自己在一个合约集群的其他节点信息
private transient CommitAlgorithm commitAlgorithm; private transient CommitAlgorithm commitAlgorithm;
private transient TrustfulExecutorConnection connection; private transient TrustfulExecutorConnection connection;
@ -137,9 +148,16 @@ public class ContractUnitController {
} }
public String getDisplayStr() { public String getDisplayStr() {
return "pSize:" + (prepare == null ? "null" : prepare.size()) + " cSize:" return "pSize:"
+ (commit == null ? "null" : commit.size()) + " isSendCommit:" + isSendCommit + (prepare == null ? "null" : prepare.size())
+ " isSendReply:" + isSendReply + " buffSize:" + buff.size(); + " cSize:"
+ (commit == null ? "null" : commit.size())
+ " isSendCommit:"
+ isSendCommit
+ " isSendReply:"
+ isSendReply
+ " buffSize:"
+ buff.size();
} }
} }
} }

20
src/main/java/org/bdware/sc/units/ContractUnitMember.java
View File

@ -4,15 +4,15 @@ import org.bdware.sc.conn.Node;
public class ContractUnitMember extends Node { public class ContractUnitMember extends Node {
private static final long serialVersionUID = 8175664649689078937L; private static final long serialVersionUID = 8175664649689078937L;
public Node node; public Node node;
public String pubKey; public String pubKey;
public ContractUnitMember(Node node) { public ContractUnitMember(Node node) {
this.node = node; this.node = node;
} }
public Node getNode() { public Node getNode() {
return node; return node;
} }
} }

153
src/main/java/org/bdware/sc/units/ContractUnitMessage.java
View File

@ -12,99 +12,98 @@ import org.zz.gmhelper.SM2Util;
public class ContractUnitMessage implements Serializable { public class ContractUnitMessage implements Serializable {
private static final long serialVersionUID = 584934384202845750L; private static final long serialVersionUID = 584934384202845750L;
ContractUnitType type; ContractUnitType type;
private String contractID; private String contractID;
String requestID; String requestID;
String signature; String signature;
byte[] content; byte[] content;
public ContractUnitMessage(String requestID, String contractID, byte[] sender) { public ContractUnitMessage(String requestID, String contractID, byte[] sender) {
this.requestID = requestID; this.requestID = requestID;
this.setContractID(contractID); this.setContractID(contractID);
} }
public ContractUnitMessage() { public ContractUnitMessage() {
// TODO Auto-generated constructor stub // TODO Auto-generated constructor stub
} }
public static ContractUnitMessage parse(byte[] b) { public static ContractUnitMessage parse(byte[] b) {
try { try {
ByteArrayInputStream bi = new ByteArrayInputStream(b); ByteArrayInputStream bi = new ByteArrayInputStream(b);
ObjectInputStream input = new ObjectInputStream(bi); ObjectInputStream input = new ObjectInputStream(bi);
ContractUnitMessage ret = (ContractUnitMessage) input.readObject(); ContractUnitMessage ret = (ContractUnitMessage) input.readObject();
return ret; return ret;
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
return null; return null;
} }
public byte[] toByteArray() { public byte[] toByteArray() {
try { try {
ByteArrayOutputStream bo = new ByteArrayOutputStream(); ByteArrayOutputStream bo = new ByteArrayOutputStream();
ObjectOutputStream output = new ObjectOutputStream(bo); ObjectOutputStream output = new ObjectOutputStream(bo);
output.writeObject(this); output.writeObject(this);
return bo.toByteArray(); return bo.toByteArray();
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
return null; return null;
} }
private static String IDA = "PBFTMsg"; private static String IDA = "PBFTMsg";
// public void sign(NodeInfo info) { // public void sign(NodeInfo info) {
// sender = info.getNodeID(); // sender = info.getNodeID();
// signature = sm02.sign(content, IDA, info.privKey).toString(); // signature = sm02.sign(content, IDA, info.privKey).toString();
// } // }
public boolean verify(ContractUnitMember member) { public boolean verify(ContractUnitMember member) {
ECPublicKeyParameters param = BCECUtil.createECPublicKeyFromStrParameters(member.pubKey, ECPublicKeyParameters param = BCECUtil.createECPublicKeyFromStrParameters(member.pubKey,SM2Util.CURVE,SM2Util.DOMAIN_PARAMS);
SM2Util.CURVE, SM2Util.DOMAIN_PARAMS);
return SM2Util.verify(param, content, signature.getBytes()); return SM2Util.verify(param,content, signature.getBytes());
} }
public void setType(ContractUnitType type) { public void setType(ContractUnitType type) {
this.type = type; this.type = type;
} }
public ContractUnitType getType() { public ContractUnitType getType() {
return type; return type;
} }
public byte[] getContent() { public byte[] getContent() {
return content; return content;
} }
public void setContent(byte[] content) { public void setContent(byte[] content) {
this.content = content; this.content = content;
} }
public String getContractID() { public String getContractID() {
return contractID; return contractID;
} }
public String getRequestID() { public String getRequestID() {
return requestID; return requestID;
} }
public String getDisplayStr() { public String getDisplayStr() {
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
sb.append(requestID); sb.append(requestID);
sb.append("_TYPE_").append(type); sb.append("_TYPE_").append(type);
sb.append("_"); sb.append("_");
sb.append(getContractID()); sb.append(getContractID());
sb.append("_"); sb.append("_");
sb.append(" content:"); sb.append(" content:");
sb.append(new String(content)); sb.append(new String(content));
return sb.toString(); return sb.toString();
} }
public void setContractID(String contractID) { public void setContractID(String contractID) {
this.contractID = contractID; this.contractID = contractID;
} }
} }

60
src/main/java/org/bdware/sc/units/ContractUnitStartRequest.java
View File

@ -10,37 +10,37 @@ import org.bdware.sc.bean.Contract;
import org.bdware.sc.conn.Node; import org.bdware.sc.conn.Node;
public class ContractUnitStartRequest implements Serializable { public class ContractUnitStartRequest implements Serializable {
/** /**
* *
*/ */
private static final long serialVersionUID = -1540780483790689627L; private static final long serialVersionUID = -1540780483790689627L;
public Contract contract; public Contract contract;
public boolean isMaster; public boolean isMaster;
public Node[] members; public Node[] members;
public static ContractUnitStartRequest parse(byte[] content) {
try {
ByteArrayInputStream input = new ByteArrayInputStream(content);
ObjectInputStream objInput;
objInput = new ObjectInputStream(input);
return (ContractUnitStartRequest) objInput.readObject();
public static ContractUnitStartRequest parse(byte[] content) { } catch (Exception e) {
try { e.printStackTrace();
ByteArrayInputStream input = new ByteArrayInputStream(content); }
ObjectInputStream objInput; return null;
objInput = new ObjectInputStream(input); }
return (ContractUnitStartRequest) objInput.readObject();
} catch (Exception e) { public byte[] toByteArray() {
e.printStackTrace(); try {
} ByteArrayOutputStream bo = new ByteArrayOutputStream();
return null; ObjectOutputStream objOutput = new ObjectOutputStream(bo);
} objOutput.writeObject(this);
return bo.toByteArray();
public byte[] toByteArray() { } catch (Exception e) {
try { e.printStackTrace();
ByteArrayOutputStream bo = new ByteArrayOutputStream(); }
ObjectOutputStream objOutput = new ObjectOutputStream(bo); return null;
objOutput.writeObject(this); }
return bo.toByteArray();
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
} }

41
src/main/java/org/bdware/sc/units/ContractUnitType.java
View File

@ -4,28 +4,27 @@ import java.io.Serializable;
public enum ContractUnitType implements Serializable { public enum ContractUnitType implements Serializable {
Start(0), AddMember(1), Sequencing(2), Unknown(3); Start(0), AddMember(1), Sequencing(2), Unknown(3);
private int type;
private int type; private ContractUnitType(int i) {
type = i;
}
private ContractUnitType(int i) { public static ContractUnitType fromByte(int i) {
type = i; switch (i) {
} case 0:
return Start;
case 1:
return AddMember;
case 2:
return Sequencing;
default:
return Unknown;
}
}
public static ContractUnitType fromByte(int i) { public int toInt() {
switch (i) { return type;
case 0: }
return Start;
case 1:
return AddMember;
case 2:
return Sequencing;
default:
return Unknown;
}
}
public int toInt() {
return type;
}
} }

18
src/main/java/org/bdware/sc/units/MultiContractMeta.java
View File

@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger;
public class MultiContractMeta implements IDSerializable { public class MultiContractMeta implements IDSerializable {
private static final Logger LOGGER = LogManager.getLogger(MultiContractMeta.class); private static final Logger LOGGER = LogManager.getLogger(MultiContractMeta.class);
private final AtomicInteger lastExeSeq; // last executed request seq private final AtomicInteger lastExeSeq; // last executed request seq
public volatile int curExeSeq = -1; // 当前正在执行请求序号 for multipoint contract requests public volatile int curExeSeq = -1; // 当前正在执行请求序号 for multipoint contract requests
public String memory; public String memory;
public String key; // privateKey public String key; // privateKey
public String publicKey; public String publicKey;
@ -101,9 +101,10 @@ public class MultiContractMeta implements IDSerializable {
public void setLastExeSeq(int lastExeSeq) { public void setLastExeSeq(int lastExeSeq) {
this.lastExeSeq.set(lastExeSeq); this.lastExeSeq.set(lastExeSeq);
if (KeyValueDBUtil.instance.containsKey(CMTables.LastExeSeq.toString(), contractID)) { // 如果现在是Stable模式就同步刷到磁盘 if (KeyValueDBUtil.instance.containsKey(
KeyValueDBUtil.instance.setValue(CMTables.LastExeSeq.toString(), contractID, CMTables.LastExeSeq.toString(), contractID)) { // 如果现在是Stable模式就同步刷到磁盘
String.valueOf(lastExeSeq)); KeyValueDBUtil.instance.setValue(
CMTables.LastExeSeq.toString(), contractID, String.valueOf(lastExeSeq));
} }
} }
@ -153,18 +154,19 @@ public class MultiContractMeta implements IDSerializable {
public String[] getMembers() { public String[] getMembers() {
return members; return members;
} }
public String joinMembers(String delimiter){
public String joinMembers(String delimiter) {
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
if (members.length > 0) if (members.length > 0)
sb.append(members[0]); sb.append(members[0]);
else else return "";
return "";
for (int i = 1; i < members.length; i++) { for (int i = 1; i < members.length; i++) {
sb.append(delimiter).append(members[i]); sb.append(delimiter).append(members[i]);
} }
return sb.toString(); return sb.toString();
} }
public void setMembers(String[] m) {
members = m;
}
public void setMembers(JsonArray members) { public void setMembers(JsonArray members) {
String[] copied = new String[members.size()]; String[] copied = new String[members.size()];

2
src/main/java/org/bdware/sc/units/PubKeyNode.java
View File

@ -3,5 +3,5 @@ package org.bdware.sc.units;
import org.bdware.sc.conn.Node; import org.bdware.sc.conn.Node;
public class PubKeyNode extends Node { public class PubKeyNode extends Node {
public String pubkey; public String pubkey;
} }

2
src/main/java/org/bdware/sc/units/RecoverFlag.java
View File

@ -1,5 +1,5 @@
package org.bdware.sc.units; package org.bdware.sc.units;
public enum RecoverFlag { public enum RecoverFlag {
Fine, ToRecover, Recovering; Fine,ToRecover,Recovering;
} }

3
src/main/java/org/bdware/sc/units/RespCache.java
View File

@ -24,8 +24,7 @@ public class RespCache {
} else { } else {
waiter.wait(5000L); waiter.wait(5000L);
timeout &= waiter.get() * 2 > count; timeout &= waiter.get() * 2 > count;
if (!timeout) if (!timeout) waiter.notifyAll();
waiter.notifyAll();
return timeout; return timeout;
} }
} }

2
src/main/java/org/bdware/sc/units/SequencingAlgorithmFactory.java
View File

@ -4,5 +4,5 @@ import org.bdware.sc.consistency.CommitAlgorithm;
import org.bdware.sc.units.ContractUnitController.ContractUnit; import org.bdware.sc.units.ContractUnitController.ContractUnit;
public interface SequencingAlgorithmFactory { public interface SequencingAlgorithmFactory {
public CommitAlgorithm create(ContractUnitStartRequest req, ContractUnit unit); public CommitAlgorithm create(ContractUnitStartRequest req, ContractUnit unit);
} }

2
src/main/java/org/bdware/sc/units/TrustfulExecutorConnection.java
View File

@ -13,4 +13,4 @@ public interface TrustfulExecutorConnection<T extends Node> {
} }
List<T> getNodes(); List<T> getNodes();
} }

10
src/main/java/org/bdware/sc/util/DHTUtil.java
View File

@ -25,8 +25,9 @@ public class DHTUtil {
String hash = HashUtil.sha3ToFixedLen(key, nodes[0].length() - 2); String hash = HashUtil.sha3ToFixedLen(key, nodes[0].length() - 2);
int l = 0, r = nodes.length - 1, m, h2l = hash.compareTo(nodes[l].substring(2)), int l = 0, r = nodes.length - 1, m,
r2h = nodes[r].substring(2).compareTo(hash), h2m; h2l = hash.compareTo(nodes[l].substring(2)), r2h = nodes[r].substring(2).compareTo(hash),
h2m;
BigInteger bigH = null; BigInteger bigH = null;
String selected; String selected;
do { do {
@ -41,7 +42,8 @@ public class DHTUtil {
break; break;
} }
if (l + 1 == r) { if (l + 1 == r) {
BigInteger bigL = getBigInteger(nodes[l]), bigR = getBigInteger(nodes[r]); BigInteger bigL = getBigInteger(nodes[l]),
bigR = getBigInteger(nodes[r]);
bigH = new BigInteger(hash, 16); bigH = new BigInteger(hash, 16);
if (bigR.subtract(bigH).compareTo(bigH.subtract(bigL)) > -1) { if (bigR.subtract(bigH).compareTo(bigH.subtract(bigL)) > -1) {
selected = nodes[l]; selected = nodes[l];
@ -63,7 +65,7 @@ public class DHTUtil {
} }
} while (true); } while (true);
if (k == 1) { if (k == 1) {
return new String[] {selected}; return new String[]{selected};
} }
List<String> ret = new ArrayList<>(); List<String> ret = new ArrayList<>();
ret.add(selected); ret.add(selected);

12
src/main/java/org/bdware/server/trustedmodel/ContractExecutor.java
View File

@ -5,16 +5,12 @@ import org.bdware.sc.conn.Node;
import org.bdware.sc.conn.OnHashCallback; import org.bdware.sc.conn.OnHashCallback;
import org.bdware.sc.conn.ResultCallback; import org.bdware.sc.conn.ResultCallback;
import java.util.Map;
public interface ContractExecutor { public interface ContractExecutor {
void execute(String requestID, ContractRequest req, ResultCallback rcb, OnHashCallback hcb); void execute(String requestID, ContractRequest req, ResultCallback rcb, OnHashCallback hcb);
default void onRecover(Map<String, Object> args) {} default void close() {
}
default void onDeliverBlock(String data) {} default void onSyncMessage(Node node, byte[] data) {
}
default void close() {}
default void onSyncMessage(Node node, byte[] data) {}
} }

3
src/main/java/org/bdware/server/trustedmodel/ContractUnitStatus.java
View File

@ -3,5 +3,6 @@ package org.bdware.server.trustedmodel;
import java.io.Serializable; import java.io.Serializable;
public enum ContractUnitStatus implements Serializable { public enum ContractUnitStatus implements Serializable {
CommonMode, StableMode; CommonMode,
StableMode;
} }

11
src/main/java/org/bdware/server/trustedmodel/MultiReqSeq.java
View File

@ -1,11 +0,0 @@
package org.bdware.server.trustedmodel;
public class MultiReqSeq {
public final int seq;
public final long startTime;
public MultiReqSeq(int s) {
seq = s;
startTime = System.currentTimeMillis();
}
}

5
src/main/java/org/bdware/server/trustedmodel/SingleNodeExecutor.java
View File

@ -14,8 +14,7 @@ public class SingleNodeExecutor implements ContractExecutor {
} }
@Override @Override
public void execute(String requestID, ContractRequest req, ResultCallback rcb, public void execute(String requestID, ContractRequest req, ResultCallback rcb, OnHashCallback hcb) {
OnHashCallback hcb) {
cm.executeLocallyAsync(req, rcb, hcb); cm.executeLocallyAsync(req, rcb, hcb);
} }
} }

2
src/test/java/org/bdware/sc/test/ContractManagerTest.java
View File

@ -2,11 +2,9 @@ package org.bdware.sc.test;
import org.bdware.sc.ContractManager; import org.bdware.sc.ContractManager;
public class ContractManagerTest { public class ContractManagerTest {
public static void main(String[] args) { public static void main(String[] args) {
ContractManager.yjsPath = "./generatedlib/yjs.jar"; ContractManager.yjsPath = "./generatedlib/yjs.jar";
ContractManager.instance = new ContractManager(); ContractManager.instance = new ContractManager();
} }
} }

20
src/test/java/org/bdware/sc/test/ReflectionTest.java
View File

@ -1,20 +0,0 @@
package org.bdware.sc.test;
import com.google.gson.Gson;
import java.lang.reflect.Method;
public class ReflectionTest {
public static class Test {
public static void go(String... args) {
System.out.println(new Gson().toJson(args));
}
}
@org.junit.Test
public void go() throws Exception {
Method m = Test.class.getDeclaredMethod("go", String[].class);
String[] abc = new String[] {"ab", "cd"};
m.invoke(null, new Object[] {abc});
}
}

27
src/test/java/org/bdware/sc/units/RespCacheTest.java
View File

@ -12,15 +12,17 @@ public class RespCacheTest {
new Thread(() -> { new Thread(() -> {
try { try {
Thread.sleep(j * 1000); Thread.sleep(j * 1000);
if (j > 2) if (j > 2) Thread.sleep(6 * 1000);
Thread.sleep(6 * 1000);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} }
boolean waitResult = cache.waitForHalf(); boolean waitResult = cache.waitForHalf();
System.out.println("tid:" + Thread.currentThread().getId() System.out.println(
+ " reach target, waitResult:" + waitResult); "tid:"
+ Thread.currentThread().getId()
+ " reach target, waitResult:"
+ waitResult);
}).start(); }).start();
} }
Thread.sleep(20000); Thread.sleep(20000);
@ -35,15 +37,17 @@ public class RespCacheTest {
new Thread(() -> { new Thread(() -> {
try { try {
Thread.sleep(j * 1000); Thread.sleep(j * 1000);
if (j > 2) if (j > 2) Thread.sleep(1900);
Thread.sleep(1900);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} }
boolean waitResult = cache.waitForHalf(); boolean waitResult = cache.waitForHalf();
System.out.println("tid:" + Thread.currentThread().getId() System.out.println(
+ " reach target, waitResult:" + waitResult); "tid:"
+ Thread.currentThread().getId()
+ " reach target, waitResult:"
+ waitResult);
}).start(); }).start();
} }
Thread.sleep(20000); Thread.sleep(20000);
@ -62,8 +66,11 @@ public class RespCacheTest {
e.printStackTrace(); e.printStackTrace();
} }
boolean waitResult = cache.waitForHalf(); boolean waitResult = cache.waitForHalf();
System.out.println("tid:" + Thread.currentThread().getId() System.out.println(
+ " reach target, waitResult:" + waitResult); "tid:"
+ Thread.currentThread().getId()
+ " reach target, waitResult:"
+ waitResult);
}).start(); }).start();
} }
Thread.sleep(10000); Thread.sleep(10000);