test/dataflow
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

加入vscode调试方式

Browse Source
This commit is contained in:
huangsimin
2022-07-27 16:25:10 +08:00
parent 4fa9035983
commit 1ea40571d9
12 changed files with 132 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/main/java/com/yuandian/dataflow/controller/PacketsProcessor.java
View File

@@ -52,40 +52,45 @@ public class PacketsProcessor implements RpcProcessor<PacketsProcessor.PacketsRe
@Override
public void handleRequest(RpcContext rpcCtx, PacketsRequest request) {
// StateServerFactory.getStateServer().updateFsmStateAsync(s, onCompleted);
var resp = new RaftResponse<>();
resp.setSuccess(true);
rpcCtx.sendResponse(resp);
var ss = StateServerFactory.getStateServer();
log.info("{} handler request.packets.size(): {}", StateServerFactory.getServerId(), request.packets.size());
log.debug("{} handler request.packets.size(): {}", StateServerFactory.getServerId(), request.packets.size());
ss.readIndexState( new GenericClosure<State>() {
@Override
public void run(Status status) {
var state = this.getValue();
var ws = state.getWorkers().get(StateServerFactory.getServerId());
ws.setTaskQueueSize( ws.getTaskQueueSize() - request.packets.size());
ws.setUpdateAt(Instant.now());
log.debug("workerState taskQueueSize: {} psize: {} state {}", ws.getTaskQueueSize(), request.packets.size(), state.getWorkers().size());
Operate.CallOperate(new Operate(OperateType.PUT,ws), new GenericClosure<Operate>() {
@Override
public void run(Status status) {
var resp = this.getResponse();
resp.setMsg(rpcCtx.getRemoteAddress());
if(status.isOk()) {
resp.setSuccess(true);
log.info("{}", resp);
} else {
resp.setSuccess(false);
log.debug("status {}", status);
if(status.isOk()) {
var state = this.getValue();
var ws = state.getWorkers().get(StateServerFactory.getServerId());
ws.setTaskQueueSize( ws.getTaskQueueSize() - request.packets.size());
ws.setUpdateAt(Instant.now());
log.debug("workerState taskQueueSize: {} psize: {} state {}", ws.getTaskQueueSize(), request.packets.size(), state.getWorkers().size());
Operate.CallOperate(new Operate(OperateType.PUT,ws), new GenericClosure<Operate>() {
@Override
public void run(Status status) {
if(status.isOk()) {
log.info("{}", resp);
}
}
rpcCtx.sendResponse(resp);
}
});
});
}
}
} );
}
@Override

82
src/main/java/com/yuandian/dataflow/statemachine/StateMachine.java
View File

@@ -20,7 +20,7 @@ import com.yuandian.dataflow.statemachine.closure.GenericClosure;
import com.yuandian.dataflow.statemachine.operate.Operate;
import com.yuandian.dataflow.statemachine.operate.Operate.OperateType;
import com.yuandian.dataflow.statemachine.rpc.RaftResponse;
import com.yuandian.dataflow.statemachine.rpc.OperateProcessor.OperateRequest;
import com.yuandian.dataflow.statemachine.state.State;
import com.yuandian.dataflow.statemachine.state.WorkerState;
@@ -38,7 +38,8 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class StateMachine extends StateMachineAdapter {
// private static final Logger LOG = LoggerFactory.getLogger(StateMachine.class);
// private static final Logger LOG =
// LoggerFactory.getLogger(StateMachine.class);
/**
* State value 全局使用的唯一状态
@@ -70,46 +71,46 @@ public class StateMachine extends StateMachineAdapter {
if (iter.done() != null) {
// This task is applied by this node, get value from closure to avoid additional
// parsing.
closure = (GenericClosure<Operate>)iter.done(); // 只支持单一个State. 全状态机只支持一种提交
closure = (GenericClosure<Operate>) iter.done(); // 只支持单一个State. 全状态机只支持一种提交
op = closure.getValue();
} else {
// Have to parse FetchAddRequest from this user log.
// Have to parse FetchAddRequest from this user log.
final ByteBuffer data = iter.getData();
try {
op = SerializerManager.getSerializer(SerializerManager.Hessian2).deserialize(data.array(),Operate.class.getName());
op = SerializerManager.getSerializer(SerializerManager.Hessian2).deserialize(
data.array(),
Operate.class.getName());
} catch (CodecException e) {
log.info("{}", e.toString());
log.info("{}", e.toString());
}
}
if(op != null) {
switch(op.getType()) {
if (op != null) {
switch (op.getType()) {
case PUT:
WorkerState ws = op.getValue();
WorkerState ws = op.getValue();
state.getWorkers().put(ws.peerId, ws);
if(closure != null) {
if (closure != null) {
closure.success(op);
closure.run(Status.OK());
}
break;
case REMOVE:
if(closure != null) {
if (closure != null) {
closure.success(op);
closure.run(Status.OK());
}
break;
default:
break;
}
} else {
}
iter.next();
}
@@ -136,21 +137,21 @@ public class StateMachine extends StateMachineAdapter {
this.leaderTerm.set(term);
// 判断是否Master线程还在跑, 如果存在则中断
if(MasterFactory.getMasterExecute().isAlive()) {
if (MasterFactory.getMasterExecute().isAlive()) {
MasterFactory.getMasterExecute().interrupt();
}
var ss = StateServerFactory.getStateServer();
ss.readIndexState( new GenericClosure<State>() {
var ss = StateServerFactory.getStateServer();
ss.readIndexState(new GenericClosure<State>() {
@Override
public void run(Status status) {
var ws = state.getWorkers().get( StateServerFactory.getServerId() );
if(ws == null) {
var ws = state.getWorkers().get(StateServerFactory.getServerId());
if (ws == null) {
ws = new WorkerState(StateServerFactory.getServerId());
}
Operate op = new Operate(OperateType.PUT, ws);
ss.applyOperate(op, new GenericClosure<Operate>() {
@Override
@@ -159,7 +160,7 @@ public class StateMachine extends StateMachineAdapter {
}
});
}
});
// 当成为master时候 必须启动
@@ -175,21 +176,17 @@ public class StateMachine extends StateMachineAdapter {
super.onLeaderStop(status);
// 判断是否Master线程还在跑, 如果存在则中断
if(MasterFactory.getMasterExecute().isAlive()) {
if (MasterFactory.getMasterExecute().isAlive()) {
MasterFactory.getMasterExecute().interrupt();
}
}
@Override
public void onShutdown() {
log.debug("onShutdown");
super.onShutdown();
}
@Override
public void onStartFollowing(LeaderChangeContext ctx) {
@@ -197,12 +194,12 @@ public class StateMachine extends StateMachineAdapter {
try {
// 判断是否Master线程还在跑, 如果存在则中断
if(MasterFactory.getMasterExecute().isAlive()) {
if (MasterFactory.getMasterExecute().isAlive()) {
MasterFactory.getMasterExecute().interrupt();
}
var ws = new WorkerState(StateServerFactory.getServerId());
log.debug("my: {} leader id {}",StateServerFactory.getServerId(), StateServerFactory.getLeaderId());
log.debug("my: {} leader id {}", StateServerFactory.getServerId(), StateServerFactory.getLeaderId());
var op = new Operate(OperateType.PUT, ws);
@@ -212,10 +209,10 @@ public class StateMachine extends StateMachineAdapter {
log.info("{} {}", status, this.getResponse());
}
});
return;
} catch (Exception e) {
log.info("{}", e.toString());
log.info("{}", e.toString());
}
super.onStartFollowing(ctx);
@@ -228,12 +225,12 @@ public class StateMachine extends StateMachineAdapter {
@Override
public void onStopFollowing(LeaderChangeContext ctx) {
log.debug("{} {}",ctx, StateServerFactory.getCluster().getServerId());
log.debug("{} {}", ctx, StateServerFactory.getCluster().getServerId());
var ss = StateServerFactory.getStateServer();
var ws = new WorkerState(StateServerFactory.getServerId());
log.debug("my: {} leader id {}",StateServerFactory.getServerId(), StateServerFactory.getLeaderId());
log.debug("my: {} leader id {}", StateServerFactory.getServerId(), StateServerFactory.getLeaderId());
var op = new Operate(OperateType.PUT, ws);
Operate.CallOperate(op, new GenericClosure() {
@@ -243,14 +240,7 @@ public class StateMachine extends StateMachineAdapter {
}
});
super.onStopFollowing(ctx);
}
}

9
src/main/java/com/yuandian/dataflow/statemachine/StateServerFactory.java
View File

@@ -188,17 +188,18 @@ public class StateServerFactory {
getNode().readIndex(BytesUtil.EMPTY_BYTES, new ReadIndexClosure() {
@Override
public void run(Status status, long index, byte[] reqCtx) {
if( status.isOk()) {
if(status.isOk()) {
// 回调失败
closure.success(ss.fsm.getState());
closure.run(status);
}
}
closure.run(status);
}
} );
}
public void applyOperate(Operate op, GenericClosure closure) {
public void applyOperate(Operate op, GenericClosure<Operate> closure) {
// 所有的提交都必须再leader进行
if (!ss.isLeader()) {
ss.handlerNotLeaderError(closure);

2
src/main/java/com/yuandian/dataflow/statemachine/operate/Operate.java
View File

@@ -2,6 +2,7 @@ package com.yuandian.dataflow.statemachine.operate;
import java.io.Serializable;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.error.RemotingException;
import com.alipay.sofa.jraft.rpc.InvokeCallback;
import com.yuandian.dataflow.statemachine.StateServerFactory;
@@ -75,6 +76,7 @@ public class Operate implements Serializable {
var resp = (RaftResponse<Operate>) result;
closure.setResponse(resp);
closure.success(resp.getValue());
closure.run(Status.OK());
}
}, 5000);

2
src/main/java/com/yuandian/dataflow/statemachine/rpc/OperateProcessor.java
View File

@@ -64,7 +64,7 @@ public class OperateProcessor implements RpcProcessor<OperateProcessor.OperateRe
log.info("request: {}", request);
final GenericClosure closure = new GenericClosure() {
final GenericClosure<Operate> closure = new GenericClosure<Operate>() {
@Override
public void run(Status status) {
rpcCtx.sendResponse(getResponse());

8
src/main/java/com/yuandian/dataflow/utils/Utils.java Normal file
View File

@@ -0,0 +1,8 @@
package com.yuandian.dataflow.utils;
public class Utils {
public static void main(String[] args) {
}
}