bdware/custom-plugin
19
0
Fork 0
mirror of https://gitee.com/BDWare/custom-plugin synced 2025-01-10 01:44:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

update mpcexecutor

Browse Source
This commit is contained in:
CaiHQ 2022-09-02 21:23:17 +08:00
parent 1356c1b56e
commit f89ad35d7a
1 changed files with 96 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
src/main/java/org/bdware/consistency/plugin/sharding/MultiPointCooperationExecutor.java
View File

@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicInteger;
public class MultiPointCooperationExecutor extends AbstractContextContractExecutor { public class MultiPointCooperationExecutor extends AbstractContextContractExecutor {
private static final Logger LOGGER = LogManager.getLogger(MultiPointCooperationExecutor.class); private static final Logger LOGGER = LogManager.getLogger(MultiPointCooperationExecutor.class);
final Object lock = new Object(); final Object lock = new Object();
private final ContractMeta meta;
int resultCount; int resultCount;
AtomicInteger request_index = new AtomicInteger(0); AtomicInteger request_index = new AtomicInteger(0);
ContractExecType type; ContractExecType type;
@ -45,6 +46,7 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
resultCount = c; resultCount = c;
contractID = con_id; contractID = con_id;
multiMeta = cmActions.getManager().multiContractRecorder.getMultiContractMeta(contractID); multiMeta = cmActions.getManager().multiContractRecorder.getMultiContractMeta(contractID);
meta = cmActions.getManager().statusRecorder.getContractMeta(contractID);
} }
public void setSeq(int seq) { public void setSeq(int seq) {
@ -69,14 +71,14 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
} }
// TODO 多调多统一个seq的有多个请求，这个需要改 // TODO 多调多统一个seq的有多个请求，这个需要改
masterServerTCPAction.getReqCache().get(contractID).put(req.seq, sendStr); masterServerTCPAction.getReqCache().get(contractID).put(req.seq, sendStr);
LOGGER.debug(JsonUtil.toJson(req)); // LOGGER.debug(JsonUtil.toJson(req));
LOGGER.info("node size = " + nodes.length); // LOGGER.info("node size = " + nodes.length);
LOGGER.debug("nodes:" + JsonUtil.toJson(nodes)); // LOGGER.debug("nodes:" + JsonUtil.toJson(nodes));
for (String node : nodes) { for (String node : nodes) {
if (node.equals(globalConf.getNodeID())) { if (node.equals(globalConf.getNodeID())) {
masterClientTCPAction.asyncExecuteContractLocally(jo, rc); masterClientTCPAction.asyncExecuteContractLocally(jo, rc);
} else { } else {
LOGGER.info("[sendRequests] get cmNode " + node.substring(0, 5) + " not null " + "RequestAllExecutor 发送请求给 " + node.substring(0, 5)); // LOGGER.info("[sendRequests] get cmNode " + node.substring(0, 5) + " not null " + "RequestAllExecutor 发送请求给 " + node.substring(0, 5));
networkManager.sendToAgent(node, sendStr); networkManager.sendToAgent(node, sendStr);
} }
} }
@ -86,7 +88,6 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
ContractClient client = cmActions.getManager().getClient(contractID); ContractClient client = cmActions.getManager().getClient(contractID);
JsonObject arg = new JsonObject(); JsonObject arg = new JsonObject();
arg.addProperty("funcName", funcName); arg.addProperty("funcName", funcName);
JsonArray funcArgs = new JsonArray(); JsonArray funcArgs = new JsonArray();
if (args != null && args.length > 0) for (JsonElement je : args) if (args != null && args.length > 0) for (JsonElement je : args)
funcArgs.add(je); funcArgs.add(je);
@ -101,39 +102,82 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
int val; int val;
if (routeInfo.useDefault == null) { if (routeInfo.useDefault == null) {
// func myFunc (requester, sourceArg) // func myFunc (requester, sourceArg)
return invokeFunctionWithoutLimit(req.getContractID(), routeInfo.funcName, String[].class, new JsonPrimitive(req.getRequester()), req.getArg()); JsonElement requester = req.getRequester() == null ? JsonNull.INSTANCE : new JsonPrimitive(req.getRequester());
return invokeFunctionWithoutLimit(req.getContractID(), routeInfo.funcName, String[].class, requester, req.getArg());
} }
switch (routeInfo.useDefault) { switch (routeInfo.useDefault) {
case byRequester: case byRequester:
val = new BigInteger(req.getRequester(), 16).mod(new BigInteger("" + members.length)).intValue(); val = new BigInteger(req.getRequester(), 16).mod(new BigInteger("" + members.length)).intValue();
while (val < 0) { while (val < 0 && members.length > 0) {
val = val + members.length; val = val + members.length;
} }
return new String[]{members[val]}; return new String[]{members[val]};
case byArgHash: case byArgHash:
val = req.getArg().hashCode(); val = req.getArg().hashCode();
val = val % members.length; val = val % members.length;
while (val < 0) { while (val < 0 && members.length > 0) {
val += members.length; val += members.length;
} }
return new String[]{members[val]}; return new String[]{members[val]};
case byTarget: case byJsonPropHash:
JsonObject jo = req.getArg().getAsJsonObject(); JsonElement jo = tryLoadJsonProp(req, routeInfo.param);
val = new BigInteger(jo.get("target").getAsString(), 16).mod(new BigInteger("" + members.length)).intValue(); val = jo.toString().hashCode() % members.length;
while (val < 0) { while (val < 0 && members.length > 0) {
val = val + members.length; val += members.length;
} }
return new String[]{members[val]}; return new String[]{members[val]};
case byShardingIDTree:
return byShardingIDTree(members, meta.contract.shardingId, Integer.valueOf(routeInfo.param));
default: default:
return members; return members;
} }
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace();
return members; return members;
} }
} }
private String[] byShardingIDTree(String[] members, int shardingId, int childCount) {
if (childCount > 0) {
int from = shardingId * childCount + 1;
int to = Math.min(members.length, shardingId * childCount + childCount + 1);
if (to > from) {
String[] ret = new String[to - from];
System.arraycopy(members, from, ret, 0, to - from);
return ret;
} else
return new String[]{};
} else {
childCount = Math.abs(childCount);
if (shardingId > 0) shardingId = (shardingId - 1) / childCount;
return new String[]{members[shardingId]};
}
}
private JsonElement tryLoadJsonProp(ContractRequest req, String param) {
try {
if (req.getArg() == null)
return JsonNull.INSTANCE;
JsonObject arg;
if (req.getArg().isJsonPrimitive()) {
arg = JsonUtil.parseString(req.getArg().getAsString()).getAsJsonObject();
} else arg = req.getArg().getAsJsonObject();
if (!param.contains(".")) {
return arg.get(param);
} else {
String[] props = param.split("\\.");
JsonElement result = arg;
for (String str : props)
result = result.getAsJsonObject().get(str);
return result;
}
} catch (Exception e) {
e.printStackTrace();
return JsonNull.INSTANCE;
}
}
public boolean checkCurNodeNumValid() { public boolean checkCurNodeNumValid() {
LOGGER.info("checkCurNodeNumValid");
String[] nodes = multiMeta.getMembers(); String[] nodes = multiMeta.getMembers();
// List<String> nodes = info.members; // List<String> nodes = info.members;
int validNode = 0; int validNode = 0;
@ -146,15 +190,14 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
} }
int c = resultCount; int c = resultCount;
if (type == ContractExecType.Sharding) c = (int) Math.ceil((double) c / 2); if (type == ContractExecType.Sharding) c = (int) Math.ceil((double) c / 2);
LOGGER.info("c=" + c + " validNode=" + validNode);
return validNode >= c; return validNode >= c;
} }
@Override @Override
public void execute(String requestID, ContractRequest req, ResultCallback rc, OnHashCallback hcb) { public void execute(String requestID, ContractRequest req, ResultCallback rc, OnHashCallback hcb) {
LOGGER.info("[MultiPointCooperationExecutor] execute " + JsonUtil.toJson(req)); //LOGGER.info("[MultiPointCooperationExecutor] execute " + JsonUtil.toJson(req));
// 获得action 函数名 // 获得action 函数名
LOGGER.info("action is : " + req.getAction()); // LOGGER.info("action is : " + req.getAction());
req.setContractID(cmActions.getManager().getContractIDByName(req.getContractID())); req.setContractID(cmActions.getManager().getContractIDByName(req.getContractID()));
if (requestID != null && requestID.endsWith("_mul")) { if (requestID != null && requestID.endsWith("_mul")) {
synchronized (lock) { synchronized (lock) {
@ -170,9 +213,9 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
} }
req.needSeq = true; req.needSeq = true;
String id = System.currentTimeMillis() + "_" + (int) (Math.random() * 1000000) + "_" + req.seq; String id = System.currentTimeMillis() + "_" + (int) (Math.random() * 1000000) + "_" + req.seq;
LOGGER.info("execute receive requestID= " + requestID + " msgID=" + id); //LOGGER.info("execute receive requestID= " + requestID + " msgID=" + id);
if (checkCurNodeNumValid()) { // 校验成功 current node num 合法 // 校验成功 current node num 合法
LOGGER.info("checkCurNodeNumValid true"); //LOGGER.info("checkCurNodeNumValid true");
ContractMeta meta = cmActions.getManager().statusRecorder.getContractMeta(req.getContractID()); ContractMeta meta = cmActions.getManager().statusRecorder.getContractMeta(req.getContractID());
FunctionDesp fun = meta.getExportedFunction(req.getAction()); FunctionDesp fun = meta.getExportedFunction(req.getAction());
ResultCallback collector; ResultCallback collector;
@ -181,22 +224,22 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
JoinInfo joinInfo = fun.joinInfo; JoinInfo joinInfo = fun.joinInfo;
RouteInfo routeInfo = fun.routeInfo; RouteInfo routeInfo = fun.routeInfo;
int count = getJoinCount(joinInfo, contractID, req); int count = getJoinCount(joinInfo, contractID, req);
LOGGER.info("requestID=" + requestID + " join Count: " + count);
String[] members = multiMeta.getMembers(); String[] members = multiMeta.getMembers();
String[] nodes = getAccordingToRouteInfo(routeInfo, req, members); String[] nodes = getAccordingToRouteInfo(routeInfo, req, members);
if (nodes.length < count) { if (nodes.length < count) {
count = nodes.length; count = nodes.length;
} }
collector = createResultCallback(id, rc, count, req.seq, req.getContractID(), joinInfo); // 初始化结果收集器 //LOGGER.info("requestID=" + requestID + " join Count: " + count + " nodeCount:" + nodes.length);
if (count > 0) {
collector = createResultCallback(id, rc, count, req.seq, req.getContractID(), joinInfo);
// 初始化结果收集器
masterServerTCPAction.getSync().sleep(id, collector); masterServerTCPAction.getSync().sleep(id, collector);
LOGGER.info("requestID=" + requestID + " master broadcasts request " + req.seq); // LOGGER.info("requestID=" + requestID + " master broadcasts request " + req.seq);
sendRequest(id, req, nodes, collector); // 发送请求 sendRequest(id, req, nodes, collector); // 发送请求
} else { } else {
LOGGER.info("invalidNodeNumOnResult"); ContractResult cr = new ContractResult(ContractResult.Status.Exception, new JsonPrimitive("broadcast 0"));
request_index.getAndDecrement(); rc.onResult(JsonUtil.toJson(cr));
ContractResult finalResult = new ContractResult(ContractResult.Status.Error, new JsonPrimitive("node number unavailbale,request refused."));
rc.onResult(JsonUtil.toJson(finalResult));
} }
} }
@ -258,15 +301,22 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
// str的data是个ContractResult // str的data是个ContractResult
// 在这儿也是返回个ContractResult // 在这儿也是返回个ContractResult
try { try {
LOGGER.info(str); //LOGGER.info(str);
JsonObject obj = JsonParser.parseString(str).getAsJsonObject(); JsonObject obj = JsonParser.parseString(str).getAsJsonObject();
String id = obj.get("nodeID").getAsString();
String id;
if (obj.get("nodeID") == null) {
id = "null";
// LOGGER.info("已经收到第" + order + "节点的结果，无节点ID！ 该结果被忽略" + order);
return;
} else
id = obj.get("nodeID").getAsString();
if (nodeIDs.contains(id)) { if (nodeIDs.contains(id)) {
LOGGER.info("已经收到节点 " + id.substring(0, 5) + " 的结果，该结果被忽略"); // LOGGER.info("已经收到节点 " + id.substring(0, 5) + " 的结果，该结果被忽略");
return; return;
} }
nodeIDs.add(id); nodeIDs.add(id);
LOGGER.info("contractID=" + contractID + " 收到第 " + order + " 个节点回复 : " + str + " order=" + order + " count=" + count); //LOGGER.info("contractID=" + contractID + " 收到第 " + order + " 个节点回复 : " + str + " order=" + order + " count=" + count);
componedContractResult.add(obj); componedContractResult.add(obj);
// 收集到所有结果 // 收集到所有结果
if (order.incrementAndGet() == count) { if (order.incrementAndGet() == count) {
@ -289,7 +339,7 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
} }
originalCallback.onResult(JsonUtil.toJson(finalResult)); originalCallback.onResult(JsonUtil.toJson(finalResult));
// } // }
LOGGER.info("本次执行最终结果为 " + finalResult.size + "个节点合并的,结果为 " + finalResult.result); //LOGGER.info("本次执行最终结果为 " + finalResult.size + "个节点合并的,结果为 " + finalResult.result);
// 集群中事务序号+1 // 集群中事务序号+1
// MasterServerTCPAction.contractID2Members.get(contractID).nextSeq(); // MasterServerTCPAction.contractID2Members.get(contractID).nextSeq();
@ -330,8 +380,8 @@ public class MultiPointCooperationExecutor extends AbstractContextContractExecut
if (joinInfo.joinFuncName != null) { if (joinInfo.joinFuncName != null) {
JsonElement ret = MultiPointCooperationExecutor.invokeFunctionWithoutLimit(contractID, joinInfo.joinFuncName, JsonElement.class, jo); JsonElement ret = MultiPointCooperationExecutor.invokeFunctionWithoutLimit(contractID, joinInfo.joinFuncName, JsonElement.class, jo);
finalResult.result = ret; finalResult.result = ret;
return;
} }
return;
} }
switch (joinInfo.useDefault) { switch (joinInfo.useDefault) {
case add: case add: