Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Refactor worker #1997

Merged
merged 3 commits into from
Feb 22, 2020
Merged

Refactor worker #1997

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6589220
Refactor worker (#2)
Technoboy- Feb 22, 2020
00c9b80
updates
Technoboy- Feb 22, 2020
39a02e6
Merge branch 'refactor-worker' into refactor-worker
Technoboy- Feb 22, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/thread/Stopper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,6 @@ public static final boolean isRunning(){
}

public static final void stop(){
signal.getAndSet(true);
signal.set(true);
}
}
92 changes: 67 additions & 25 deletions ...cheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.dolphinscheduler.remote.codec.NettyDecoder;
import org.apache.dolphinscheduler.remote.codec.NettyEncoder;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
import org.apache.dolphinscheduler.remote.exceptions.RemotingException;
import org.apache.dolphinscheduler.remote.exceptions.RemotingTimeoutException;
Expand All @@ -33,7 +34,8 @@
import org.apache.dolphinscheduler.remote.future.ReleaseSemaphore;
import org.apache.dolphinscheduler.remote.future.ResponseFuture;
import org.apache.dolphinscheduler.remote.handler.NettyClientHandler;
import org.apache.dolphinscheduler.remote.utils.Address;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.CallerThreadExecutePolicy;
import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory;
import org.slf4j.Logger;
Expand Down Expand Up @@ -64,7 +66,7 @@ public class NettyRemotingClient {
/**
* channels
*/
private final ConcurrentHashMap<Address, Channel> channels = new ConcurrentHashMap(128);
private final ConcurrentHashMap<Host, Channel> channels = new ConcurrentHashMap(128);

/**
* started flag
Expand Down Expand Up @@ -158,17 +160,17 @@ public void run() {

/**
* async send
* @param address address
* @param host host
* @param command command
* @param timeoutMillis timeoutMillis
* @param invokeCallback callback function
* @throws InterruptedException
* @throws RemotingException
*/
public void sendAsync(final Address address, final Command command,
public void sendAsync(final Host host, final Command command,
final long timeoutMillis,
final InvokeCallback invokeCallback) throws InterruptedException, RemotingException {
final Channel channel = getChannel(address);
final Channel channel = getChannel(host);
if (channel == null) {
throw new RemotingException("network error");
}
Expand Down Expand Up @@ -214,7 +216,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
});
} catch (Throwable ex){
responseFuture.release();
throw new RemotingException(String.format("send command to address: %s failed", address), ex);
throw new RemotingException(String.format("send command to host: %s failed", host), ex);
}
} else{
String message = String.format("try to acquire async semaphore timeout: %d, waiting thread num: %d, total permits: %d",
Expand All @@ -225,17 +227,17 @@ public void operationComplete(ChannelFuture future) throws Exception {

/**
* sync send
* @param address address
* @param host host
* @param command command
* @param timeoutMillis timeoutMillis
* @return command
* @throws InterruptedException
* @throws RemotingException
*/
public Command sendSync(final Address address, final Command command, final long timeoutMillis) throws InterruptedException, RemotingException {
final Channel channel = getChannel(address);
public Command sendSync(final Host host, final Command command, final long timeoutMillis) throws InterruptedException, RemotingException {
final Channel channel = getChannel(host);
if (channel == null) {
throw new RemotingException(String.format("connect to : %s fail", address));
throw new RemotingException(String.format("connect to : %s fail", host));
}
final long opaque = command.getOpaque();
final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, null, null);
Expand All @@ -250,7 +252,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
}
responseFuture.setCause(future.cause());
responseFuture.putResponse(null);
logger.error("send command {} to address {} failed", command, address);
logger.error("send command {} to host {} failed", command, host);
}
});
/**
Expand All @@ -259,49 +261,89 @@ public void operationComplete(ChannelFuture future) throws Exception {
Command result = responseFuture.waitResponse();
if(result == null){
if(responseFuture.isSendOK()){
throw new RemotingTimeoutException(address.toString(), timeoutMillis, responseFuture.getCause());
throw new RemotingTimeoutException(host.toString(), timeoutMillis, responseFuture.getCause());
} else{
throw new RemotingException(address.toString(), responseFuture.getCause());
throw new RemotingException(host.toString(), responseFuture.getCause());
}
}
return result;
}

public void send(final Host host, final Command command) throws RemotingException {
Channel channel = getChannel(host);
if (channel == null) {
throw new RemotingException(String.format("connect to : %s fail", host));
}
try {
ChannelFuture future = channel.writeAndFlush(command).await();
if (future.isSuccess()) {
logger.debug("send command : {} , to : {} successfully.", command, host.getAddress());
} else {
String msg = String.format("send command : %s , to :%s failed", command, host.getAddress());
logger.error(msg, future.cause());
throw new RemotingException(msg);
}
} catch (Exception e) {
logger.error("Send command {} to address {} encounter error.", command, host.getAddress());
throw new RemotingException(String.format("Send command : %s , to :%s encounter error", command, host.getAddress()), e);
}
}

/**
* register processor
* @param commandType command type
* @param processor processor
*/
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor) {
this.registerProcessor(commandType, processor, null);
}

/**
* register processor
*
* @param commandType command type
* @param processor processor
* @param executor thread executor
*/
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor, final ExecutorService executor) {
this.clientHandler.registerProcessor(commandType, processor, executor);
}

/**
* get channel
* @param address
* @param host
* @return
*/
public Channel getChannel(Address address) {
Channel channel = channels.get(address);
public Channel getChannel(Host host) {
Channel channel = channels.get(host);
if(channel != null && channel.isActive()){
return channel;
}
return createChannel(address, true);
return createChannel(host, true);
}

/**
* create channel
* @param address address
* @param host host
* @param isSync sync flag
* @return channel
*/
public Channel createChannel(Address address, boolean isSync) {
public Channel createChannel(Host host, boolean isSync) {
ChannelFuture future;
try {
synchronized (bootstrap){
future = bootstrap.connect(new InetSocketAddress(address.getHost(), address.getPort()));
future = bootstrap.connect(new InetSocketAddress(host.getIp(), host.getPort()));
}
if(isSync){
future.sync();
}
if (future.isSuccess()) {
Channel channel = future.channel();
channels.put(address, channel);
channels.put(host, channel);
return channel;
}
} catch (Exception ex) {
logger.info("connect to {} error {}", address, ex);
logger.info("connect to {} error {}", host, ex);
}
return null;
}
Expand Down Expand Up @@ -341,10 +383,10 @@ private void closeChannels(){

/**
* close channel
* @param address address
* @param host host
*/
public void closeChannel(Address address){
Channel channel = this.channels.remove(address);
public void closeChannel(Host host){
Channel channel = this.channels.remove(host);
if(channel != null){
channel.close();
}
Expand Down
2 changes: 1 addition & 1 deletion ...emote/src/main/java/org/apache/dolphinscheduler/remote/command/ExecuteTaskAckCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */package org.apache.dolphinscheduler.remote.command;import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;import java.io.Serializable;import java.util.Date;/** * execute task request command */public class ExecuteTaskAckCommand implements Serializable { private int taskInstanceId; private Date startTime; private String host; private int status; private String logPath; private String executePath; public Date getStartTime() { return startTime; } public void setStartTime(Date startTime) { this.startTime = startTime; } public String getHost() { return host; } public void setHost(String host) { this.host = host; } public int getStatus() { return status; } public void setStatus(int status) { this.status = status; } public int getTaskInstanceId() { return taskInstanceId; } public void setTaskInstanceId(int taskInstanceId) { this.taskInstanceId = taskInstanceId; } public String getLogPath() { return logPath; } public void setLogPath(String logPath) { this.logPath = logPath; } public String getExecutePath() { return executePath; } public void setExecutePath(String executePath) { this.executePath = executePath; } /** * package request command * * @return command */ public Command convert2Command(long opaque){ Command command = new Command(opaque); command.setType(CommandType.EXECUTE_TASK_ACK); byte[] body = FastJsonSerializer.serialize(this); command.setBody(body); return command; } @Override public String toString() { return "ExecuteTaskAckCommand{" + "taskInstanceId=" + taskInstanceId + ", startTime=" + startTime + ", host='" + host + '\'' + ", status=" + status + ", logPath='" + logPath + '\'' + ", executePath='" + executePath + '\'' + '}'; }}
/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */package org.apache.dolphinscheduler.remote.command;import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;import java.io.Serializable;import java.util.Date;/** * execute task request command */public class ExecuteTaskAckCommand implements Serializable { private int taskInstanceId; private Date startTime; private String host; private int status; private String logPath; private String executePath; public Date getStartTime() { return startTime; } public void setStartTime(Date startTime) { this.startTime = startTime; } public String getHost() { return host; } public void setHost(String host) { this.host = host; } public int getStatus() { return status; } public void setStatus(int status) { this.status = status; } public int getTaskInstanceId() { return taskInstanceId; } public void setTaskInstanceId(int taskInstanceId) { this.taskInstanceId = taskInstanceId; } public String getLogPath() { return logPath; } public void setLogPath(String logPath) { this.logPath = logPath; } public String getExecutePath() { return executePath; } public void setExecutePath(String executePath) { this.executePath = executePath; } /** * package request command * * @return command */ public Command convert2Command(){ Command command = new Command(); command.setType(CommandType.EXECUTE_TASK_ACK); byte[] body = FastJsonSerializer.serialize(this); command.setBody(body); return command; } @Override public String toString() { return "ExecuteTaskAckCommand{" + "taskInstanceId=" + taskInstanceId + ", startTime=" + startTime + ", host='" + host + '\'' + ", status=" + status + ", logPath='" + logPath + '\'' + ", executePath='" + executePath + '\'' + '}'; }}
Expand Down
2 changes: 1 addition & 1 deletion ...e/src/main/java/org/apache/dolphinscheduler/remote/command/ExecuteTaskRequestCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */package org.apache.dolphinscheduler.remote.command;import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;import java.io.Serializable;/** * execute task request command */public class ExecuteTaskRequestCommand implements Serializable { /** * task instance json */ private String taskInfoJson; public String getTaskInfoJson() { return taskInfoJson; } public void setTaskInfoJson(String taskInfoJson) { this.taskInfoJson = taskInfoJson; } public ExecuteTaskRequestCommand() { } public ExecuteTaskRequestCommand(String taskInfoJson) { this.taskInfoJson = taskInfoJson; } /** * package request command * * @return command */ public Command convert2Command(){ Command command = new Command(); command.setType(CommandType.EXECUTE_TASK_REQUEST); byte[] body = FastJsonSerializer.serialize(this); command.setBody(body); return command; } @Override public String toString() { return "ExecuteTaskRequestCommand{" + "taskInfoJson='" + taskInfoJson + '\'' + '}'; }}
/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */package org.apache.dolphinscheduler.remote.command;import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;import java.io.Serializable;/** * execute task request command */public class ExecuteTaskRequestCommand implements Serializable { /** * task execution context */ private String taskExecutionContext; public String getTaskExecutionContext() { return taskExecutionContext; } public void setTaskExecutionContext(String taskExecutionContext) { this.taskExecutionContext = taskExecutionContext; } public ExecuteTaskRequestCommand() { } public ExecuteTaskRequestCommand(String taskExecutionContext) { this.taskExecutionContext = taskExecutionContext; } /** * package request command * * @return command */ public Command convert2Command(){ Command command = new Command(); command.setType(CommandType.EXECUTE_TASK_REQUEST); byte[] body = FastJsonSerializer.serialize(this); command.setBody(body); return command; } @Override public String toString() { return "ExecuteTaskRequestCommand{" + "taskExecutionContext='" + taskExecutionContext + '\'' + '}'; }}
Expand Down
Loading