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

TODO: ok

Browse Source
This commit is contained in:
huangsimin 2022-08-01 23:25:01 +08:00
parent cbf0308841
commit c339e6e3f0
5 changed files with 14 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/main/java/com/yuandian/dataflow/controller/MasterProcessor.java
View File

@ -69,7 +69,10 @@ public class MasterProcessor implements MasterExecute {
} }
var canTasks = MAX_TASKS - ws.getTaskQueueSize(); var canTasks = MAX_TASKS - ws.getTaskQueueSize();
log.info("剩余能处理的任务数量[{}] :{}", peer, canTasks); if(canTasks > 0) {
log.info("剩余能处理的任务数量[{}] :{}", peer, canTasks);
}
if (canTasks <= 0) { if (canTasks <= 0) {
return; return;
} }
@ -93,19 +96,18 @@ public class MasterProcessor implements MasterExecute {
new GenericClosure() { new GenericClosure() {
@Override @Override
public void run(Status status) { public void run(Status status) {
log.info("PacketsRequest run {}", status); // log.info("PacketsRequest run {}", status);
try { try {
StateFactory.rpcClientInvokeAsync(peer.getEndpoint(), request, StateFactory.rpcClientInvokeAsync(peer.getEndpoint(), request,
new InvokeCallback() { new InvokeCallback() {
@Override @Override
public void complete(Object result, Throwable err) { public void complete(Object result, Throwable err) {
latch.countDown(); latch.countDown();
log.debug("countDown {}", latch.getCount());
if (err != null) { if (err != null) {
// TODO: 如果错误, 需要让节点恢复任务处理的状态 // TODO: 如果错误, 需要让节点恢复任务处理的状态
log.debug("{}", err); log.debug("{}", err);
} }
log.debug("PacketsRequest: {}", result); // log.debug("PacketsRequest: {}", result);
} }
}, 5000); }, 5000);
} catch (InterruptedException | RemotingException e) { } catch (InterruptedException | RemotingException e) {

4
src/main/java/com/yuandian/dataflow/statemachine/StateFactory.java
View File

@ -237,10 +237,10 @@ public class StateFactory {
public void readIndexState(GenericClosure closure) { public void readIndexState(GenericClosure closure) {
getNode().readIndex(BytesUtil.EMPTY_BYTES, new ReadIndexClosure(2000) { getNode().readIndex(BytesUtil.EMPTY_BYTES, new ReadIndexClosure(5000) {
@Override @Override
public void run(Status status, long index, byte[] reqCtx) { public void run(Status status, long index, byte[] reqCtx) {
log.debug("readIndexState({}) {}", getServerId(), status); // log.debug("readIndexState({}) {}", getServerId(), status);
if (status.isOk()) { if (status.isOk()) {
closure.success(ss.fsm.getState()); closure.success(ss.fsm.getState());
closure.setValue(ss.fsm.getState()); closure.setValue(ss.fsm.getState());

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

@ -91,13 +91,13 @@ public class StateMachine extends StateMachineAdapter {
case PUT_WORKERSTATE: case PUT_WORKERSTATE:
WorkerState ws = op.getValue(); WorkerState ws = op.getValue();
log.debug("PUT {}", ws.peerId); // log.debug("PUT {}", ws.peerId);
state.getWorkers().put(ws.peerId, ws); state.getWorkers().put(ws.peerId, ws);
break; break;
case GET_STATE: case GET_STATE:
closure.setValue(this.state); closure.setValue(this.state);
log.info("GET_STATE value={} at logIndex={}", this.state, iter.getIndex()); // log.info("GET_STATE value={} at logIndex={}", this.state, iter.getIndex());
break; break;
case REMOVE: case REMOVE:

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

@ -63,7 +63,7 @@ public class Operate implements Serializable {
*/ */
@java.lang.SuppressWarnings("unchecked") @java.lang.SuppressWarnings("unchecked")
public static void CallOperate(Operate op, GenericClosure closure) { public static void CallOperate(Operate op, GenericClosure closure) {
log.debug("CallOperate Value {}", op.<WorkerState>getValue()); // log.debug("CallOperate Value {}", op.<WorkerState>getValue());
// 如果是leader 就直接提交 // 如果是leader 就直接提交
if (StateFactory.isLeader()) { if (StateFactory.isLeader()) {
StateFactory.applyOperate(op, closure); StateFactory.applyOperate(op, closure);
@ -78,7 +78,7 @@ public class Operate implements Serializable {
StateFactory.rpcClientInvokeAsync(StateFactory.getLeaderId().getEndpoint(), request, new InvokeCallback() { StateFactory.rpcClientInvokeAsync(StateFactory.getLeaderId().getEndpoint(), request, new InvokeCallback() {
@Override @Override
public void complete(Object result, Throwable err) { public void complete(Object result, Throwable err) {
log.debug("Object result {}", result); // log.debug("Object result {}", result);
var resp = (RaftResponse) result; var resp = (RaftResponse) result;
closure.setResponse(resp); closure.setResponse(resp);
closure.success(resp.getValue()); closure.success(resp.getValue());

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

@ -54,14 +54,14 @@ public class OperateProcessor implements RpcProcessor<OperateProcessor.OperateRe
@Override @Override
public void handleRequest(RpcContext rpcCtx, OperateRequest request) { public void handleRequest(RpcContext rpcCtx, OperateRequest request) {
log.info("request: {}", request); // log.info("request: {}", request);
final GenericClosure closure = new GenericClosure () { final GenericClosure closure = new GenericClosure () {
@Override @Override
public void run(Status status) { public void run(Status status) {
if(status.isOk()) { if(status.isOk()) {
log.info("{}", status); // log.info("{}", status);
rpcCtx.sendResponse(getResponse()); rpcCtx.sendResponse(getResponse());
return; return;
} }