home/pinpoint
2
0
Fork 0
mirror of https://github.com/wahyd4/pinpoint.git synced 2026-08-18 01:06:03 +10:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #19 from nstopkimsk/master

Browse Source 
modified comments in English
This commit is contained in:
Sungkwan Kim
2014-12-28 18:15:19 +09:00
parent a9ab50991a a70bf4535c
commit 7df00a611e
14 changed files with 100 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocket.java
+11 -7
View File
@@ -69,7 +69,7 @@ public class PinpointSocket {
}
logger.warn("reconnectSocketHandler:{}", socketHandler);
// Pinpoint 소켓 내부 객체가 되기전에 listener를 먼저 등록
// register listener before becoming internal object of Pinpoint socket.
socketHandler.doHandshake();
this.socketHandler = socketHandler;
@@ -77,8 +77,11 @@ public class PinpointSocket {
notifyReconnectEvent();
}
// reconnectEventListener의 경우 직접 생성자 호출시에 Dummy를 포함하고 있으며,
// setter를 통해서도 접근을 못하게 하기 때문에 null이 아닌 것이 보장됨
/*
because reconnectEventListener's constructor contains Dummy and can't be access through setter,
guarantee it is not null.
*/
public boolean addPinpointSocketReconnectEventListener(PinpointSocketReconnectEventListener eventListener) {
if (eventListener == null) {
return false;
@@ -125,8 +128,8 @@ public class PinpointSocket {
}
public ClientStreamChannelContext createStreamChannel(byte[] payload, ClientStreamChannelMessageListener clientStreamChannelMessageListener) {
// 실패를 리턴하는 StreamChannel을 던져야 되는데. StreamChannel을 interface로 변경해야 됨.
// 일단 그냥 ex를 던지도록 하겠음.
// StreamChannel must be changed into interface in order to throw the StreamChannel that returns failure.
// fow now throw just exception
ensureOpen();
return socketHandler.createStreamChannel(payload, clientStreamChannelMessageListener);
}
@@ -150,8 +153,9 @@ public class PinpointSocket {
}
/**
* ping packet을 tcp 채널에 write한다.
* write 실패시 PinpointSocketException이 throw 된다.
* write pinng packet on tcp channel
* PinpointSocketException throws when writing fails.
*
*/
public void sendPing() {
SocketHandler socketHandler = this.socketHandler;
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactory.java
+27 -16
View File
@@ -73,8 +73,9 @@ public class PinpointSocketFactory {
private long reconnectDelay = 3 * 1000;
private final Timer timer;
// 이 값이 짧아야 될 필요가 없음. client에서 server로 가는 핑 주기를 짧게 유지한다고 해서.
// 연결끊김이 빨랑 디텍트 되는게 아님. 오히려 server에서 client의 ping주기를 짧게 해야 디텍트 속도가 빨라짐.
// it's better to be a long value. even though keeping ping period from client to server short,
// disconnection between them dose not be detected quickly.
// rather keeping it from server to client short help detect disconnection as soon as possible.
private long pingDelay = DEFAULT_PING_DELAY;
private long enableWorkerPacketDelay = DEFAULT_ENABLE_WORKER_PACKET_DELAY;
private long timeoutMillis = DEFAULT_TIMEOUTMILLIS;
@@ -94,7 +95,8 @@ public class PinpointSocketFactory {
if (bossCount < 1) {
throw new IllegalArgumentException("bossCount is negative: " + bossCount);
}
// timer를 connect timeout으로 쓰므로 먼저 만들어야 됨.
// create a timer earlier because it is used for connectTimeout
Timer timer = createTimer();
ClientBootstrap bootstrap = createBootStrap(bossCount, workerCount, timer);
setOptions(bootstrap);
@@ -118,12 +120,13 @@ public class PinpointSocketFactory {
private void setOptions(ClientBootstrap bootstrap) {
// connectTimeout
bootstrap.setOption(CONNECT_TIMEOUT_MILLIS, DEFAULT_CONNECT_TIMEOUT);
// read write timeout이 있어야 되나? nio라서 없어도 되던가?
// read write timeout이 있어야 되나?
// read write timeout needed? isn't it needed because of nio?
// tcp 세팅
// tcp setting
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("keepAlive", true);
// buffer
// buffer setting
bootstrap.setOption("sendBufferSize", 1024 * 64);
bootstrap.setOption("receiveBufferSize", 1024 * 64);
@@ -219,10 +222,13 @@ public class PinpointSocketFactory {
traceSocket(pinpointSocket);
return pinpointSocket;
}
/*
trace mechanism is needed in case of calling close without closing socket
it is okay to make that later because this is a exceptional case.
*/
private void traceSocket(PinpointSocket pinpointSocket) {
// socket을 닫지 않고 clsoe했을 경우의 추적 로직이 필요함
// 예외 케이스 이므로 나중에 만들어도 될듯.
}
public PinpointSocket scheduledConnect(String host, int port) {
@@ -318,7 +324,8 @@ public class PinpointSocketFactory {
if (timeout.isCancelled()) {
return;
}
// 이벤트는 fire됬지만 close됬을 경우 reconnect를 시도 하지 않음.
// Just return not to try reconnection when event has been fired but pinpointSocket already closed.
if (pinpointSocket.isClosed()) {
logger.debug("pinpointSocket is already closed.");
return;
@@ -341,11 +348,14 @@ public class PinpointSocketFactory {
pinpointSocket.reconnectSocketHandler(socketHandler);
} else {
if (!pinpointSocket.isClosed()) {
// 구지 여기서 안찍어도 exceptionCought에서 메시지가 발생하므로 생략
// if (logger.isWarnEnabled()) {
// Throwable cause = future.getCause();
// logger.warn("reconnect fail. {} Caused:{}", socketAddress, cause.getMessage());
// }
/*
// comment out because exception message can be taken at exceptionCought
if (logger.isWarnEnabled()) {
Throwable cause = future.getCause();
logger.warn("reconnect fail. {} Caused:{}", socketAddress, cause.getMessage());
}
*/
reconnect(pinpointSocket, socketAddress);
} else {
logger.info("pinpointSocket is closed. stop reconnect.");
@@ -372,7 +382,8 @@ public class PinpointSocketFactory {
if (!stop.isEmpty()) {
logger.info("stop Timeout:{}", stop.size());
}
// stop 뭔가 취소를 해야 되나??
// stop, cancel something?
}
Map<String, Object> getProperties() {
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java
+15 -13
View File
@@ -375,10 +375,10 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke
} else {
boolean cancel = channelFuture.cancel();
if (cancel) {
// 3초에도 io가 안끝나면 일단 timeout인가?
// if IO not finished in 3 seconds, dose it mean timeout?
throw new PinpointSocketException("io timeout");
} else {
// 성공했으니. 위와 로직이 동일할듯.
// same logic as above because of success
boolean success = channelFuture.isSuccess();
if (success) {
return;
@@ -448,7 +448,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke
case PacketType.APPLICATION_RESPONSE:
this.requestManager.messageReceived((ResponsePacket) message, e.getChannel());
return;
// connector로 들어오는 request 메시지를 핸들링을 해야 함.
// have to handle a request message through connector
case PacketType.APPLICATION_REQUEST:
this.messageListener.handleRequest((RequestPacket) message, e.getChannel());
return;
@@ -481,7 +481,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke
private void messageReceivedServerClosed(Channel channel) {
logger.info("ServerClosed Packet received. {}", channel);
// reconnect 상태로 변경한다.
state.setState(State.RECONNECT);
}
@@ -523,13 +523,14 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
Throwable cause = e.getCause();
if (state.getState() == State.INIT_RECONNECT) {
// 재접속시 stackTrace는 제거하였음. 로그가 너무 많이 나옴.
logger.info("exceptionCaught() reconnect fail. state:{} {} Caused:{}", state.getString(), e.getChannel(), cause.getMessage());
// removed stackTrace when reconnect. so many logs.
logger.info("exceptionCaught() reconnect failed. state:{} {} Caused:{}", state.getString(), e.getChannel(), cause.getMessage());
} else {
logger.warn("exceptionCaught() UnexpectedError happened. state:{} {} Caused:{}", state.getString(), e.getChannel(), cause.getMessage(), cause);
}
// error가 발생하였을 경우의 동작을 더 정확히 해야 될듯함.
// 아래처럼 하면 상대방이 그냥 죽었을때 reconnet가 안됨.
// need to handle a error more precisely.
// below code dose not reconnect when node on channel is just hang up or dead without specific reasons.
// state.setClosed();
// Channel channel = e.getChannel();
// if (channel.isConnected()) {
@@ -547,7 +548,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke
} else if(currentState == State.INIT_RECONNECT){
logger.debug("channelClosed() reconnect fail. state:{} {}", state.getString(currentState), e.getChannel());
} else if (state.isRun(currentState) || currentState == State.RECONNECT) {
// 여기서 부터 비정상 closed라고 볼수 있다.
// abnormal closed from here
if (state.isRun(currentState)) {
logger.debug("change state=reconnect");
state.setState(State.RECONNECT);
@@ -600,9 +601,10 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke
logger.debug("close() state change complete");
// hand shake close
final Channel channel = this.channel;
// close packet을 먼저 날리고 resource를 정리해야 되나?
// resource 정리시 request response 메시지에 대한 에러 처리나, stream 채널의 정리가 필요하니 반대로 해야 되나?? 이게 맞는거 같긴한데. timer가 헤깔리네.
// 헤깔리니. 일단 만들고 추후 수정.
// is it correct that send a "close packet" first and release resources?
// when you release resources, you need to clear messages about request/response and stream channel. need to handle reversely?
// handling timer is unclear so just make and enhance later.
sendClosedPacket(channel);
releaseResource();
logger.debug("channel.close()");
@@ -639,7 +641,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
logger.warn("ClientClosePacket write fail. channel:{}", future.getCause(), future.getCause());
logger.warn("ClientClosePacket write failed. channel:{}", future.getCause(), future.getCause());
} else {
logger.debug("ClientClosePacket write success. channel:{}", future.getChannel());
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketReconnectEventListener.java
+5 -3
View File
@@ -17,9 +17,11 @@
package com.navercorp.pinpoint.rpc.client;
public interface PinpointSocketReconnectEventListener {
// 현재는 Reconnect를 제외한 별다른 Event가 없음
// 이후에 별다른 Event가 있을 경우 Event와 함께 넘겨주면 좋을듯함
/*
there is no event except "reconnect" currently.
when additional events are needed, it will be useful to pass with Event
*/
void reconnectPerformed(PinpointSocket socket);
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/RequestManager.java
+6 -5
View File
@@ -42,7 +42,7 @@ public class RequestManager {
private final AtomicInteger requestId = new AtomicInteger(1);
private final ConcurrentMap<Integer, DefaultFuture<ResponseMessage>> requestMap = new ConcurrentHashMap<Integer, DefaultFuture<ResponseMessage>>();
// Timer를 factory로 옮겨야 되나?
// Have to move Timer into factory?
private final Timer timer;
@@ -61,7 +61,7 @@ public class RequestManager {
public boolean fireFailure() {
DefaultFuture<ResponseMessage> future = removeMessageFuture(requestId);
if (future != null) {
// 정확하게 지워짐.
// removed perfectly.
return true;
}
return false;
@@ -79,7 +79,7 @@ public class RequestManager {
Timeout timeout = timer.newTimeout(future, timeoutMillis, TimeUnit.MILLISECONDS);
future.setTimeout(timeout);
} catch (IllegalStateException e) {
// timer가 shutdown되었을 경우인데. 이것은 socket이 closed되었다는 의미뿐이 없을거임..
// this case is that timer has been shutdown. That maybe just means that socket has been closed.
future.setFailure(new PinpointSocketException("socket closed")) ;
}
}
@@ -125,7 +125,8 @@ public class RequestManager {
if (old != null) {
throw new PinpointSocketException("unexpected error. old future exist:" + old + " id:" + requestId);
}
// future가 실패하였을 경우 requestMap에서 빠르게 지울수 있도록 핸들을 넣는다.
// when future fails, put a handle in order to remove a failed future in the requestMap.
FailureEventHandler removeTable = createFailureEventHandler(requestId);
future.setFailureEventHandler(removeTable);
@@ -138,7 +139,7 @@ public class RequestManager {
logger.debug("close()");
final PinpointSocketException closed = new PinpointSocketException("socket closed");
// close의 동시성 타이밍을 좀더 좋게 맞출수는 없나?
// Could you handle race conditions of "close" more precisely?
// final Timer timer = this.timer;
// if (timer != null) {
// Set<Timeout> stop = timer.stop();
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/State.java
+3 -3
View File
@@ -29,16 +29,16 @@ public class State {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
// 프로파일러에서 동작하는 것들은 최대한 가볍게 함
// operations on profiler should be light as much possible
// 0 핸드쉐이크 안함.. 1은 동작중, 2는 closed
// 0 : no handshake, 1: running
public static final int INIT_RECONNECT = -1;
public static final int INIT = 0;
public static final int RUN = 1;
public static final int RUN_DUPLEX_COMMUNICATION = 2;
public static final int RUN_SIMPLEX_COMMUNICATION = 3;
public static final int CLOSED = 4;
// 이 상태가 있어야 되나?
// need this state?
public static final int RECONNECT = 5;
rpc/src/main/java/com/navercorp/pinpoint/rpc/codec/PacketDecoder.java
+5 -4
View File
@@ -87,12 +87,12 @@ public class PacketDecoder extends FrameDecoder {
case PacketType.CONTROL_PING:
readPing(packetType, buffer);
sendPong(channel);
// 그냥 ping은 버리자.
// just drop ping
return null;
case PacketType.CONTROL_PONG:
logger.debug("receive pong. {}", channel);
readPong(packetType, buffer);
// pong 도 그냥 버리자.
// just also drop pong.
return null;
case PacketType.CONTROL_HANDSHAKE:
return readEnableWorker(packetType, buffer);
@@ -105,8 +105,9 @@ public class PacketDecoder extends FrameDecoder {
}
private void sendPong(Channel channel) {
// ping에 대한 응답으로 pong은 자동으로 응답한다.
logger.debug("receive ping. send pong. {}", channel);
// a "pong" responds to a "ping" automatically.
logger.debug("received ping. sending pong. {}", channel);
ChannelFuture write = channel.write(PongPacket.PONG_PACKET);
write.addListener(pongWriteFutureListener);
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/control/ControlMessageEncoder.java
+4 -2
View File
@@ -27,12 +27,14 @@ import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
/**
be simple. this is similar to NPC but use "bit" operation instead of Chunk in String.
permit only utf-8 encoding.
* @author koo.taejin
*/
public class ControlMessageEncoder {
// 단순하게 가자 NPC랑 비슷 단) String에서 Chunk대신 bit 연산 사용하게 함
// UTF-8만 사용하게 함
private Charset charset;
public ControlMessageEncoder() {
rpc/src/main/java/com/navercorp/pinpoint/rpc/packet/PacketType.java
+2 -2
View File
@@ -44,11 +44,11 @@ public class PacketType {
public static final short CONTROL_CLIENT_CLOSE = 100;
public static final short CONTROL_SERVER_CLOSE = 110;
// 컨트롤 패킷
// control packet
public static final short CONTROL_HANDSHAKE = 150;
public static final short CONTROL_HANDSHAKE_RESPONSE = 151;
// ping, pong의 경우 성능상 두고 다른 CONTROL은 이걸로 뺌
// keep stay because of performance in case of ping and pong. others removed.
public static final short CONTROL_PING = 200;
public static final short CONTROL_PONG = 201;
rpc/src/main/java/com/navercorp/pinpoint/rpc/packet/PayloadPacket.java
+1 -2
View File
@@ -52,11 +52,10 @@ public class PayloadPacket {
public static ChannelBuffer appendPayload(final ChannelBuffer header, final byte[] payload) {
if (payload == null) {
// 이건 payload 헤더이긴하다.
// this is also payload header
header.writeInt(-1);
return header;
} else {
// 이건 payload 헤더이긴하다.
header.writeInt(payload.length);
ChannelBuffer payloadWrap = ChannelBuffers.wrappedBuffer(payload);
return ChannelBuffers.wrappedBuffer(true, header, payloadWrap);
rpc/src/main/java/com/navercorp/pinpoint/rpc/server/PinpointServerSocket.java
+12 -12
View File
@@ -157,13 +157,13 @@ public class PinpointServerSocket extends SimpleChannelHandler {
}
private void setOptions(ServerBootstrap bootstrap) {
// read write timeout이 있어야 되나? nio라서 없어도 되던가?
// write timeout은 별도 interceptor를 통해서 이루어 져야 함. write timeout은 있음.
// is read/write timeout necessary? don't need it because of NIO?
// write timeout should be set through additional interceptor. write timeout exists.
// tcp 세팅
// tcp setting
bootstrap.setOption("child.tcpNoDelay", true);
bootstrap.setOption("child.keepAlive", true);
// buffer
// buffer setting
bootstrap.setOption("child.sendBufferSize", 1024 * 64);
bootstrap.setOption("child.receiveBufferSize", 1024 * 64);
@@ -185,7 +185,7 @@ public class PinpointServerSocket extends SimpleChannelHandler {
}
private ServerBootstrap createBootStrap(int bossCount, int workerCount) {
// profiler, collector,
// profiler, collector
ExecutorService boss = Executors.newCachedThreadPool(new PinpointThreadFactory("Pinpoint-Server-Boss"));
NioServerBossPool nioServerBossPool = new NioServerBossPool(boss, bossCount, ThreadNameDeterminer.CURRENT);
@@ -291,9 +291,9 @@ public class PinpointServerSocket extends SimpleChannelHandler {
logger.debug("received ClientClosePacket {}", channel);
ChannelContext channelContext = getChannelContext(channel);
channelContext.changeStateBeingShutdown();
// 상대방이 닫는거에 반응해서 socket을 닫도록 하자.
// channel.close();
// close socket when the node on channel close socket
// channel.close();
}
private void handleStreamPacket(StreamPacket packet, Channel channel) {
@@ -404,8 +404,8 @@ public class PinpointServerSocket extends SimpleChannelHandler {
super.channelDisconnected(ctx, e);
}
// 참고 ChannelClose 이벤트는 상대방이 먼저 연결을 끊어 Disconnected가 발생했을 경우에도 발생이 가능함
// 이부분 염두하고 코드 작성이 필요함
// ChannelClose event may also happen when the other party close socket first and Disconnected occurs
// Should consider that.
@Override
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
final Channel channel = e.getChannel();
@@ -520,7 +520,7 @@ public class PinpointServerSocket extends SimpleChannelHandler {
logger.debug("newPingTimeout");
pingTimer.newTimeout(pintTask, 1000 * 60 * 5, TimeUnit.MILLISECONDS);
} catch (IllegalStateException e) {
// timer가 stop일 경우 정지.
// stop in case of timer stopped
logger.debug("timer stopped. Caused:{}", e.getMessage());
}
}
@@ -547,7 +547,7 @@ public class PinpointServerSocket extends SimpleChannelHandler {
bootstrap = null;
}
// 요청을 죽인뒤에 timer를 제거함
// clear the request first and remove timer
requestManagerTimer.stop();
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/server/PinpointServerSocketState.java
+3 -3
View File
@@ -37,13 +37,13 @@ public class PinpointServerSocketState {
this.currentState = state;
return true;
} else if (PinpointServerSocketStateCode.isFinished(this.currentState)) {
// 상태가 더 이상 변경할수 없는 것들은 로그만 출력
// 이미 종료 상태이기 때문에 이렇게 처리해도 큰 문제가 없음
// if state can't be changed, just log.
// no problem because the state of socket has been already closed.
PinpointServerSocketStateCode checkBefore = this.beforeState;
PinpointServerSocketStateCode checkCurrent = this.currentState;
String errorMessage = cannotChangeMessage(checkBefore, checkCurrent, state);
this.beforeState = this.currentState;
this.currentState = PinpointServerSocketStateCode.ERROR_ILLEGAL_STATE_CHANGE;
rpc/src/main/java/com/navercorp/pinpoint/rpc/server/PinpointServerSocketStateCode.java
+4 -5
View File
@@ -24,11 +24,10 @@ import java.util.Set;
*/
public enum PinpointServerSocketStateCode {
// 상태는 다음과 같다.
// NONE : 아무 이벤트가 없는 상태
// RUN : 서버로 메시지를 보내는 것만 가능한 Socket 상태
// RUN_DUPLEX_COMMUNICATION : 양방향 통신이 가능한 Socket 상태 (서버로 메시지를 보내는 것 서버가 보낸 메시지를 처리하는 것 )
// BEING_SHUTDOWN : CLOSE 등의 명령을 받고 연결을 종료를 대기하는 상태
// NONE : No event
// RUN : can send message only to server
// RUN_DUPLEX_COMMUNICATION : can communicate each other by full-duplex
// BEING_SHUTDOWN : waiting to close connection CLOSE 등의 명령을 받고 연결을 종료를 대기하는 상태
// SHUTDOWN : 내가 끊거나 종료대기가 되어있는 상태일때 종료
// UNEXPECTED_SHUTDOWN : CLOSE 등의 명령을 받지 못한 상태에서 상대방이 연결을 종료하였을떄
rpc/src/main/java/com/navercorp/pinpoint/rpc/server/ServerMessageListener.java
+2 -1
View File
@@ -28,7 +28,8 @@ import com.navercorp.pinpoint.rpc.packet.SendPacket;
public interface ServerMessageListener {
void handleSend(SendPacket sendPacket, SocketChannel channel);
// 외부 노출 Channel은 별도의 Tcp Channel로 감싸는걸로 변경할 것.
// TODO make another tcp channel in case of exposed channel.
void handleRequest(RequestPacket requestPacket, SocketChannel channel);
HandshakeResponseCode handleHandshake(Map properties);