You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/04/15 17:09:49 UTC
[GitHub] [iotdb] yschengzi commented on a diff in pull request #5557: [IOTDB-2933] detect sender exit and set pipestatus=STOP
yschengzi commented on code in PR #5557:
URL: https://github.com/apache/iotdb/pull/5557#discussion_r851376891
##########
server/src/main/java/org/apache/iotdb/db/sync/receiver/ReceiverService.java:
##########
@@ -143,34 +152,47 @@ public SyncResponse receiveMsg(SyncRequest request) {
/** create and start a new pipe named pipeName */
private void createPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("create Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- createDir(pipeName, remoteIp, createTime);
- receiverManager.createPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
+ if (pipeInfo == null || pipeInfo.getStatus().equals(PipeStatus.DROP)) {
+ logger.info(
+ "create Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
+ createDir(pipeName, remoteIp, createTime);
+ receiverManager.createPipe(pipeName, remoteIp, createTime);
+ }
}
/** start an existed pipe named pipeName */
private void startPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("start Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- receiverManager.startPipe(pipeName, remoteIp);
- collector.startPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
+ if (pipeInfo != null && pipeInfo.getStatus().equals(PipeStatus.STOP)) {
+ logger.info("start Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
+ receiverManager.startPipe(pipeName, remoteIp);
+ collector.startPipe(pipeName, remoteIp, createTime);
+ }
}
/** stop an existed pipe named pipeName */
private void stopPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("stop Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- receiverManager.stopPipe(pipeName, remoteIp);
- collector.stopPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
Review Comment:
the same reason as what happens in createPipe.
##########
server/src/main/java/org/apache/iotdb/db/sync/receiver/ReceiverService.java:
##########
@@ -143,34 +152,47 @@ public SyncResponse receiveMsg(SyncRequest request) {
/** create and start a new pipe named pipeName */
private void createPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("create Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- createDir(pipeName, remoteIp, createTime);
- receiverManager.createPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
Review Comment:
pipeName and remoteIp is not enough for judge, consider following situation:
1. user create a pipe at sender, but receiver returen ACK for CREATE SyncRequest error, but the PipeInfo remains in receiver.
2. user try again with the same pipeName and remoteIp, at this time, it works normally, but due to this line, a new pipe will not be created but keep the pipe created last with wront createTime.
##########
server/src/main/java/org/apache/iotdb/db/sync/receiver/ReceiverService.java:
##########
@@ -143,34 +152,47 @@ public SyncResponse receiveMsg(SyncRequest request) {
/** create and start a new pipe named pipeName */
private void createPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("create Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- createDir(pipeName, remoteIp, createTime);
- receiverManager.createPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
+ if (pipeInfo == null || pipeInfo.getStatus().equals(PipeStatus.DROP)) {
+ logger.info(
+ "create Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
+ createDir(pipeName, remoteIp, createTime);
+ receiverManager.createPipe(pipeName, remoteIp, createTime);
+ }
}
/** start an existed pipe named pipeName */
private void startPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("start Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- receiverManager.startPipe(pipeName, remoteIp);
- collector.startPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
+ if (pipeInfo != null && pipeInfo.getStatus().equals(PipeStatus.STOP)) {
+ logger.info("start Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
+ receiverManager.startPipe(pipeName, remoteIp);
+ collector.startPipe(pipeName, remoteIp, createTime);
+ }
}
/** stop an existed pipe named pipeName */
private void stopPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("stop Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- receiverManager.stopPipe(pipeName, remoteIp);
- collector.stopPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
+ if (pipeInfo != null && pipeInfo.getStatus().equals(PipeStatus.RUNNING)) {
+ logger.info("stop Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
+ receiverManager.stopPipe(pipeName, remoteIp);
+ collector.stopPipe(pipeName, remoteIp, createTime);
+ }
}
/** drop an existed pipe named pipeName */
private void dropPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("drop Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- receiverManager.dropPipe(pipeName, remoteIp);
- collector.stopPipe(pipeName, remoteIp, createTime);
- PipeDataQueueFactory.removeBufferedPipeDataQueue(
- SyncPathUtil.getReceiverPipeLogDir(pipeName, remoteIp, createTime));
- File dir = new File(SyncPathUtil.getReceiverPipeDir(pipeName, remoteIp, createTime));
- FileUtils.deleteDirectory(dir);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
Review Comment:
the same reason as what happens in createPipe.
##########
server/src/main/java/org/apache/iotdb/db/sync/receiver/ReceiverService.java:
##########
@@ -143,34 +152,47 @@ public SyncResponse receiveMsg(SyncRequest request) {
/** create and start a new pipe named pipeName */
private void createPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("create Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- createDir(pipeName, remoteIp, createTime);
- receiverManager.createPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
+ if (pipeInfo == null || pipeInfo.getStatus().equals(PipeStatus.DROP)) {
+ logger.info(
+ "create Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
+ createDir(pipeName, remoteIp, createTime);
+ receiverManager.createPipe(pipeName, remoteIp, createTime);
+ }
}
/** start an existed pipe named pipeName */
private void startPipe(String pipeName, String remoteIp, long createTime) throws IOException {
- logger.info("start Pipe name={}, remoteIp={}, createTime={}", pipeName, remoteIp, createTime);
- receiverManager.startPipe(pipeName, remoteIp);
- collector.startPipe(pipeName, remoteIp, createTime);
+ PipeInfo pipeInfo = receiverManager.getPipeInfo(pipeName, remoteIp);
Review Comment:
the same reason as what happens in createPipe.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org