You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2022/03/27 11:47:31 UTC
[iotdb] branch new_sync updated: [To new_sync][IOTDB-1907] implement customized sync process: sender 3 (#5159)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch new_sync
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/new_sync by this push:
new a1f7336 [To new_sync][IOTDB-1907] implement customized sync process: sender 3 (#5159)
a1f7336 is described below
commit a1f7336957f806b544fe6a570ee3163bd1c66120
Author: yschengzi <87...@users.noreply.github.com>
AuthorDate: Sun Mar 27 19:46:47 2022 +0800
[To new_sync][IOTDB-1907] implement customized sync process: sender 3 (#5159)
---
.../db/integration/sync/IoTDBSyncSenderIT.java | 275 ++++++++++++++++-
.../db/integration/sync/TransportClientMock.java | 67 +++++
.../src/test/resources/iotdb-engine.properties | 3 +-
pom.xml | 2 +-
.../org/apache/iotdb/db/concurrent/ThreadName.java | 3 +-
.../org/apache/iotdb/db/conf/IoTDBConstant.java | 1 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 2 +
.../apache/iotdb/db/newsync/conf/SyncConstant.java | 14 +-
.../apache/iotdb/db/newsync/conf/SyncPathUtil.java | 25 ++
.../db/newsync/pipedata/DeletionPipeData.java | 16 -
.../apache/iotdb/db/newsync/pipedata/PipeData.java | 2 -
.../iotdb/db/newsync/pipedata/SchemaPipeData.java | 6 -
.../iotdb/db/newsync/pipedata/TsFilePipeData.java | 24 +-
.../pipedata/queue/BufferedPipeDataQueue.java | 46 +--
.../db/newsync/pipedata/queue/PipeDataQueue.java | 2 +
.../iotdb/db/newsync/receiver/ReceiverService.java | 48 +--
.../db/newsync/sender/pipe/IoTDBPipeSink.java | 8 +
.../apache/iotdb/db/newsync/sender/pipe/Pipe.java | 7 +-
.../db/newsync/sender/pipe/TransportHandler.java | 136 +++++++++
.../iotdb/db/newsync/sender/pipe/TsFilePipe.java | 229 ++++++--------
.../newsync/sender/recovery/SenderLogAnalyzer.java | 22 +-
.../db/newsync/sender/recovery/SenderLogger.java | 15 +
.../sender/recovery/TsFilePipeLogAnalyzer.java | 179 -----------
.../newsync/sender/recovery/TsFilePipeLogger.java | 224 ++------------
.../db/newsync/sender/service/SenderService.java | 134 +++++++--
.../newsync/transport/client/ITransportClient.java | 8 +-
.../newsync/transport/client/TransportClient.java | 330 ++++++++++++++-------
.../transport/conf/TransportSenderConfig.java | 129 ++++++++
.../transport/server/TransportServerManager.java | 5 +-
.../transport/server/TransportServiceImpl.java | 13 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 15 +-
.../apache/iotdb/db/utils/EnvironmentUtils.java | 2 +
.../pipedata/BufferedPipeDataQueueTest.java | 2 +-
.../db/newsync/transport/TransportServiceTest.java | 20 +-
server/src/test/resources/iotdb-engine.properties | 1 +
35 files changed, 1259 insertions(+), 756 deletions(-)
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncSenderIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncSenderIT.java
index 9a36b9b..cc92fe4 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncSenderIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncSenderIT.java
@@ -18,20 +18,287 @@
*/
package org.apache.iotdb.db.integration.sync;
-import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.newsync.pipedata.PipeData;
+import org.apache.iotdb.db.newsync.pipedata.SchemaPipeData;
+import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
+import org.apache.iotdb.db.newsync.sender.pipe.TransportHandler;
+import org.apache.iotdb.db.newsync.sender.pipe.TsFilePipe;
+import org.apache.iotdb.db.newsync.sender.service.SenderService;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.itbase.category.LocalStandaloneTest;
+import org.apache.iotdb.jdbc.Config;
+import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
+import org.junit.Test;
import org.junit.experimental.categories.Category;
-import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.util.List;
@Category({LocalStandaloneTest.class})
public class IoTDBSyncSenderIT {
+ private boolean enableSeqSpaceCompaction;
+ private boolean enableUnseqSpaceCompaction;
+ private boolean enableCrossSpaceCompaction;
+
+ private static final String pipeSinkName = "test_pipesink";
+ private static final String pipeName = "test_pipe";
+
+ private TsFilePipe pipe;
+ private TransportClientMock mock;
+
@Before
- public void setUp() throws StorageEngineException, IOException {
- EnvironmentUtils.cleanEnv();
+ public void setUp() throws Exception {
EnvironmentUtils.envSetUp();
+ enableSeqSpaceCompaction =
+ IoTDBDescriptor.getInstance().getConfig().isEnableSeqSpaceCompaction();
+ enableUnseqSpaceCompaction =
+ IoTDBDescriptor.getInstance().getConfig().isEnableUnseqSpaceCompaction();
+ enableCrossSpaceCompaction =
+ IoTDBDescriptor.getInstance().getConfig().isEnableCrossSpaceCompaction();
+ IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(false);
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqSpaceCompaction(false);
+ IoTDBDescriptor.getInstance().getConfig().setEnableCrossSpaceCompaction(false);
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+ EnvironmentUtils.shutdownDaemon();
+ EnvironmentUtils.cleanEnv();
+ }
+
+ private void prepareSchema() throws Exception {
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("set storage group to root.sg1");
+ statement.execute("set storage group to root.sg2");
+ statement.execute("create timeseries root.sg1.d1.s1 with datatype=int32, encoding=PLAIN");
+ statement.execute("create timeseries root.sg1.d1.s2 with datatype=float, encoding=RLE");
+ statement.execute("create timeseries root.sg1.d1.s3 with datatype=TEXT, encoding=PLAIN");
+ statement.execute("create timeseries root.sg1.d2.s4 with datatype=int64, encoding=PLAIN");
+ statement.execute("create timeseries root.sg2.d1.s0 with datatype=int32, encoding=PLAIN");
+ statement.execute("create timeseries root.sg2.d2.s1 with datatype=boolean, encoding=PLAIN");
+ }
+ }
+
+ private void prepareIns1() throws Exception { // add one seq tsfile in sg1
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("insert into root.sg1.d1(timestamp, s1, s2, s3) values(1, 1, 16.0, 'a')");
+ statement.execute("insert into root.sg1.d1(timestamp, s1, s2, s3) values(2, 2, 25.16, 'b')");
+ statement.execute("insert into root.sg1.d1(timestamp, s1, s2, s3) values(3, 3, 65.25, 'c')");
+ statement.execute(
+ "insert into root.sg1.d1(timestamp, s1, s2, s3) values(16, 25, 100.0, 'd')");
+ statement.execute("insert into root.sg1.d2(timestamp, s4) values(1, 1)");
+ statement.execute("flush");
+ }
+ }
+
+ private void prepareIns2() throws Exception { // add one seq tsfile in sg1
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "insert into root.sg1.d1(timestamp, s1, s2, s3) values(100, 65, 16.25, 'e')");
+ statement.execute(
+ "insert into root.sg1.d1(timestamp, s1, s2, s3) values(65, 100, 25.0, 'f')");
+ statement.execute("insert into root.sg1.d2(timestamp, s4) values(200, 100)");
+ statement.execute("flush");
+ }
+ }
+
+ private void prepareIns3()
+ throws
+ Exception { // add one seq tsfile in sg1, one unseq tsfile in sg1, one seq tsfile in sg2
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("insert into root.sg2.d1(timestamp, s0) values(100, 100)");
+ statement.execute("insert into root.sg2.d1(timestamp, s0) values(65, 65)");
+ statement.execute("insert into root.sg2.d2(timestamp, s1) values(1, true)");
+ statement.execute(
+ "insert into root.sg1.d1(timestamp, s1, s2, s3) values(25, 16, 65.16, 'g')");
+ statement.execute(
+ "insert into root.sg1.d1(timestamp, s1, s2, s3) values(200, 100, 16.65, 'h')");
+ statement.execute("flush");
+ }
+ }
+
+ private void preparePipeAndSetMock() throws Exception {
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("start pipeserver");
+ statement.execute("create pipesink " + pipeSinkName + " as iotdb");
+ statement.execute("create pipe " + pipeName + " to " + pipeSinkName);
+ pipe = (TsFilePipe) SenderService.getInstance().getRunningPipe();
+ mock =
+ new TransportClientMock(SenderService.getInstance().getRunningPipe(), "127.0.0.1", 2333);
+ TransportHandler handler =
+ new TransportHandler(
+ mock, pipeName, SenderService.getInstance().getRunningPipe().getCreateTime());
+ ((TsFilePipe) SenderService.getInstance().getRunningPipe()).setTransportHandler(handler);
+ Thread.sleep(1000L);
+ statement.execute("stop pipeserver");
+ }
+ }
+
+ private void startPipe() throws Exception {
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("start pipe " + pipeName);
+ }
+ }
+
+ private void stopPipe() throws Exception {
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("stop pipe " + pipeName);
+ }
+ }
+
+ private void dropPipe() throws Exception {
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("drop pipe " + pipeName);
+ }
+ }
+
+ private void checkResult(List<PipeData> list) { // check ins1, ins2, ins3
+ Assert.assertEquals(list.size(), 13);
+ for (int i = 0; i < 8; i++) {
+ Assert.assertTrue(list.get(i) instanceof SchemaPipeData);
+ }
+ for (int i = 9; i < list.size(); i++) {
+ Assert.assertTrue(list.get(i) instanceof TsFilePipeData);
+ }
+ }
+
+ @Test
+ public void testHistoryInsert() {
+ try {
+ prepareSchema();
+ prepareIns1();
+ prepareIns2();
+ prepareIns3();
+
+ preparePipeAndSetMock();
+ startPipe();
+ Thread.sleep(1000L);
+ checkResult(mock.getPipeDataList());
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ } finally {
+ try {
+ dropPipe();
+ Thread.sleep(1000L);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testHistoryAndRealTimeInsert() {
+ try {
+ prepareSchema();
+ prepareIns1();
+ prepareIns2();
+
+ preparePipeAndSetMock();
+ startPipe();
+ Thread.sleep(1000L);
+ prepareIns3();
+ Thread.sleep(1000L);
+ checkResult(mock.getPipeDataList());
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ } finally {
+ try {
+ dropPipe();
+ Thread.sleep(1000L);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testStopAndStart() {
+ try {
+ prepareSchema();
+ prepareIns1();
+
+ preparePipeAndSetMock();
+ startPipe();
+ prepareIns2();
+ stopPipe();
+ prepareIns3();
+ startPipe();
+ Thread.sleep(1000L);
+ checkResult(mock.getPipeDataList());
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ } finally {
+ try {
+ dropPipe();
+ Thread.sleep(1000L);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testRealTimeSchemaAndStop() {
+ try {
+ preparePipeAndSetMock();
+ prepareSchema();
+ startPipe();
+ prepareIns1();
+ stopPipe();
+ prepareIns2();
+ startPipe();
+ prepareIns3();
+ stopPipe();
+
+ Thread.sleep(1000L);
+ checkResult(mock.getPipeDataList());
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ } finally {
+ try {
+ dropPipe();
+ Thread.sleep(1000L);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
}
}
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/sync/TransportClientMock.java b/integration/src/test/java/org/apache/iotdb/db/integration/sync/TransportClientMock.java
new file mode 100644
index 0000000..a48a0a7
--- /dev/null
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/sync/TransportClientMock.java
@@ -0,0 +1,67 @@
+/*
+ * 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.iotdb.db.integration.sync;
+
+import org.apache.iotdb.db.exception.SyncConnectionException;
+import org.apache.iotdb.db.newsync.pipedata.PipeData;
+import org.apache.iotdb.db.newsync.sender.pipe.Pipe;
+import org.apache.iotdb.db.newsync.transport.client.ITransportClient;
+import org.apache.iotdb.service.transport.thrift.ResponseType;
+import org.apache.iotdb.service.transport.thrift.SyncRequest;
+import org.apache.iotdb.service.transport.thrift.SyncResponse;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class TransportClientMock implements ITransportClient {
+ private final Pipe pipe;
+ private final String ipAddress;
+ private final int port;
+
+ private List<PipeData> pipeDataList;
+
+ public TransportClientMock(Pipe pipe, String ipAddress, int port) {
+ this.pipe = pipe;
+ this.ipAddress = ipAddress;
+ this.port = port;
+
+ this.pipeDataList = new ArrayList<>();
+ }
+
+ @Override
+ public SyncResponse heartbeat(SyncRequest syncRequest) throws SyncConnectionException {
+ return new SyncResponse(ResponseType.INFO, "");
+ }
+
+ @Override
+ public void run() {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ PipeData pipeData = pipe.take();
+ pipeDataList.add(pipeData);
+ pipe.commit();
+ }
+ } catch (InterruptedException e) {
+ }
+ }
+
+ public List<PipeData> getPipeDataList() {
+ return pipeDataList;
+ }
+}
diff --git a/integration/src/test/resources/iotdb-engine.properties b/integration/src/test/resources/iotdb-engine.properties
index f93226c..43e8182 100644
--- a/integration/src/test/resources/iotdb-engine.properties
+++ b/integration/src/test/resources/iotdb-engine.properties
@@ -17,4 +17,5 @@
base_dir=target/tmp
data_dirs=target/data
-wal_dir=target/wal
\ No newline at end of file
+wal_dir=target/wal
+newsync_dir=target/newsync
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index a84d8a9..c70ad31 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1052,7 +1052,7 @@
</activation>
<properties>
<os.classifier>windows-x86_64</os.classifier>
- <thrift.download-url>http://artfiles.org/apache.org/thrift/${thrift.version}/thrift-${thrift.version}.exe</thrift.download-url>
+ <thrift.download-url>http://archive.apache.org/dist/thrift/${thrift.version}/thrift-${thrift.version}.exe</thrift.download-url>
<thrift.executable>thrift-${thrift.version}-win-x86_64.exe</thrift.executable>
<thrift.skip-making-executable>true</thrift.skip-making-executable>
<thrift.exec-cmd.executable>echo</thrift.exec-cmd.executable>
diff --git a/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java b/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
index c407012..bc4008c 100644
--- a/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
+++ b/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
@@ -56,7 +56,8 @@ public enum ThreadName {
TIMED_FlUSH_UNSEQ_MEMTABLE("Timed-Flush-Unseq-Memtable"),
TIMED_CLOSE_TSFILE("Timed-Close-TsFile"),
SETTLE_SERVICE("Settle"),
- PIPE_SERVICE("Pipe"),
+ SYNC_SENDER_PIPE("Sync-Pipe"),
+ SYNC_SENDER_HEARTBEAT("Sync-Heartbeat"),
SYNC_RECEIVER_COLLECTOR("Sync-Collector"),
CONTINUOUS_QUERY_SERVICE("ContinuousQueryTaskPoolManager"),
CLUSTER_INFO_SERVICE("ClusterInfoClient"),
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
index f0f939c..18dc87f 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
@@ -130,6 +130,7 @@ public class IoTDBConstant {
public static final String COLUMN_PIPE_CREATE_TIME = "create time";
public static final String COLUMN_PIPE2PIPESINK_NAME = "pipeSink";
public static final String COLUMN_PIPE_STATUS = "status";
+ public static final String COLUMN_PIPE_MSG = "message";
// sync receiver
public static final String COLUMN_PIPE_REMOTE_IP = "remote ip";
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 2c3a80b..ae739c6 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -230,6 +230,8 @@ public class IoTDBDescriptor {
conf.setWalDir(properties.getProperty("wal_dir", conf.getWalDir()));
+ conf.setNewSyncDir(properties.getProperty("newsync_dir", conf.getNewSyncDir()));
+
int mlogBufferSize =
Integer.parseInt(
properties.getProperty(
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncConstant.java b/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncConstant.java
index 316fcaa..5dd39f5 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncConstant.java
@@ -19,6 +19,9 @@
package org.apache.iotdb.db.newsync.conf;
public class SyncConstant {
+ /** common */
+ public static final String PIPE_LOG_DIR_NAME = "pipe-log";
+
/** sender */
public static final String DEFAULT_PIPE_SINK_IP = "127.0.0.1";
@@ -26,15 +29,19 @@ public class SyncConstant {
public static final String SENDER_PIPE_DIR_NAME = "sender";
public static final String FINISH_COLLECT_LOCK_NAME = "finishCollect.lock";
+ public static final String HISTORY_PIPE_LOG_DIR_NAME = "history-" + PIPE_LOG_DIR_NAME;
+ public static final Long DEFAULT_HEARTBEAT_DELAY_SECONDS = 10 * 60L;
public static final Long DEFAULT_WAITING_FOR_TSFILE_CLOSE_MILLISECONDS = 500L;
public static final Long DEFAULT_WAITING_FOR_TSFILE_RETRY_NUMBER = 10L;
public static final Long DEFAULT_WAITING_FOR_DEREGISTER_MILLISECONDS = 100L;
+ public static final Long DEFAULT_WAITTING_FOR_STOP_MILLISECONDS = 100L;
public static final String MODS_OFFSET_FILE_SUFFIX = ".offset";
- public static final String PIPE_LOG_NAME_SUFFIX = "-pipe.log";
+ public static final String PIPE_LOG_NAME_SUFFIX = "_pipe.log";
public static final Long DEFAULT_PIPE_LOG_SIZE_IN_BYTE = 10485760L;
- public static final String HISTORY_PIPE_LOG_NAME = PIPE_LOG_NAME_SUFFIX + ".history";
public static final String COMMIT_LOG_NAME = "commit.log";
+ public static final String UNKNOWN_IP = "UNKNOWN";
+
public static final String SENDER_LOG_NAME = "senderService.log";
public static final String PLAN_SERIALIZE_SPLIT_CHARACTER = ",";
public static final String SENDER_LOG_SPLIT_CHARACTER = " ";
@@ -44,7 +51,7 @@ public class SyncConstant {
}
public static Long getSerialNumberFromPipeLogName(String pipeLogName) {
- return Long.parseLong(pipeLogName.split("-")[0]);
+ return Long.parseLong(pipeLogName.split("_")[0]);
}
/** receiver */
@@ -53,6 +60,5 @@ public class SyncConstant {
public static final String RECEIVER_DIR = "receiver";
public static final String RECEIVER_LOG_NAME = "receiverService.log";
public static final String RECEIVER_MSG_LOG_NAME = "receiverMessage.log";
- public static final String PIPE_LOG_DIR_NAME = "pipe-log";
public static final String FILE_DATA_DIR_NAME = "file-data";
}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java b/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java
index 143a7f9..a7a4c1c 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java
@@ -19,8 +19,10 @@
package org.apache.iotdb.db.newsync.conf;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.qp.utils.DatetimeUtils;
import java.io.File;
+import java.io.IOException;
/** Util for path generation in sync module */
public class SyncPathUtil {
@@ -37,6 +39,16 @@ public class SyncPathUtil {
+ String.format("-%s-%d", pipeName, createTime);
}
+ public static String getSenderHistoryPipeDataDir(String pipeName, long createTime) {
+ return getSenderPipeDir(pipeName, createTime)
+ + File.separator
+ + SyncConstant.HISTORY_PIPE_LOG_DIR_NAME;
+ }
+
+ public static String getSenderRealTimePipeDataDir(String pipeName, long createTime) {
+ return getSenderPipeDir(pipeName, createTime) + File.separator + SyncConstant.PIPE_LOG_DIR_NAME;
+ }
+
/** receiver */
public static String getReceiverPipeLogDir(String pipeName, String remoteIp, long createTime) {
return getReceiverPipeDir(pipeName, remoteIp, createTime)
@@ -72,4 +84,17 @@ public class SyncPathUtil {
+ File.separator
+ SyncConstant.SYNC_SYS_DIR;
}
+
+ /** common */
+ public static boolean createFile(File file) throws IOException {
+ if (!file.getParentFile().exists()) {
+ file.getParentFile().mkdirs();
+ }
+ return file.createNewFile();
+ }
+
+ public static String createMsg(String msg) {
+ return String.format(
+ "[%s] %s", DatetimeUtils.convertLongToDate(DatetimeUtils.currentTime()), msg);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/DeletionPipeData.java b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/DeletionPipeData.java
index 1813b34..28a9226 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/DeletionPipeData.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/DeletionPipeData.java
@@ -27,7 +27,6 @@ import org.apache.iotdb.db.newsync.receiver.load.ILoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -66,21 +65,6 @@ public class DeletionPipeData extends PipeData {
}
@Override
- public void sendToTransport() {
- try (ByteArrayOutputStream bytesStream = new ByteArrayOutputStream();
- DataOutputStream stream = new DataOutputStream(bytesStream)) {
- serialize(stream);
- // senderTransport(bytesStream.toArray, this);
- // System.out.println(this);
- } catch (IOException e) {
- logger.warn(
- String.format(
- "Serialize deletion pipeData %s error, can not send to transport, because %s.",
- this, e));
- }
- }
-
- @Override
public String toString() {
return "DeletionData{" + "deletion=" + deletion + ", serialNumber=" + serialNumber + '}';
}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/PipeData.java b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/PipeData.java
index c8d9389..e093df1 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/PipeData.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/PipeData.java
@@ -88,8 +88,6 @@ public abstract class PipeData {
public abstract ILoader createLoader();
- public abstract void sendToTransport();
-
public enum Type {
TSFILE,
DELETION,
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/SchemaPipeData.java b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/SchemaPipeData.java
index 53164d2..52c5309 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/SchemaPipeData.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/SchemaPipeData.java
@@ -79,12 +79,6 @@ public class SchemaPipeData extends PipeData {
}
@Override
- public void sendToTransport() {
- // senderTransport(getBytes(), this);
- // System.out.println(this);
- }
-
- @Override
public String toString() {
return "SchemaPipeData{" + "serialNumber=" + serialNumber + ", plan=" + plan + '}';
}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java
index 115af1a..acfb485 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java
@@ -49,7 +49,13 @@ public class TsFilePipeData extends PipeData {
String sep = File.separator.equals("\\") ? "\\\\" : File.separator;
String[] paths = tsFilePath.split(sep);
tsFileName = paths[paths.length - 1];
- parentDirPath = tsFilePath.substring(0, tsFilePath.length() - tsFileName.length());
+ if (paths.length > 1) {
+ parentDirPath =
+ tsFilePath.substring(
+ 0, tsFilePath.length() - tsFileName.length() - File.separator.length());
+ } else {
+ parentDirPath = "";
+ }
}
public TsFilePipeData(String parentDirPath, String tsFileName, long serialNumber) {
@@ -94,15 +100,7 @@ public class TsFilePipeData extends PipeData {
return new TsFileLoader(new File(getTsFilePath()));
}
- @Override
- public void sendToTransport() {
- if (waitForTsFileClose()) {
- // senderTransprot(getFiles(), this);
- System.out.println(this);
- }
- }
-
- public List<File> getTsFiles() throws FileNotFoundException {
+ public List<File> getTsFiles(boolean shouldWaitForTsFileClose) throws FileNotFoundException {
File tsFile = new File(getTsFilePath()).getAbsoluteFile();
File resource = new File(tsFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX);
File mods = new File(tsFile.getAbsolutePath() + ModificationFile.FILE_SUFFIX);
@@ -114,6 +112,12 @@ public class TsFilePipeData extends PipeData {
files.add(tsFile);
if (resource.exists()) {
files.add(resource);
+ } else {
+ if (shouldWaitForTsFileClose && !waitForTsFileClose()) {
+ throw new FileNotFoundException(
+ String.format(
+ "Can not find %s, maybe the tsfile is not closed yet", resource.getAbsolutePath()));
+ }
}
if (mods.exists()) {
files.add(mods);
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/BufferedPipeDataQueue.java b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/BufferedPipeDataQueue.java
index 62020fd..3e73fec 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/BufferedPipeDataQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/BufferedPipeDataQueue.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.db.newsync.pipedata.queue;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.newsync.conf.SyncConstant;
+import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
import org.apache.iotdb.db.newsync.pipedata.PipeData;
import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
import org.apache.iotdb.db.utils.FileUtils;
@@ -73,6 +74,7 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
this.pipeLogStartNumber = new LinkedBlockingDeque<>();
this.outputDeque = new LinkedBlockingDeque<>();
+ this.pullSerialNumber = Long.MIN_VALUE;
this.commitSerialNumber = Long.MIN_VALUE;
recover();
@@ -95,7 +97,7 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
List<Long> startNumbers = new ArrayList<>();
for (File file : logDir.listFiles())
- if (file.getName().endsWith(SyncConstant.PIPE_LOG_NAME_SUFFIX)) {
+ if (file.getName().endsWith(SyncConstant.PIPE_LOG_NAME_SUFFIX) && file.length() > 0) {
startNumbers.add(SyncConstant.getSerialNumberFromPipeLogName(file.getName()));
}
if (startNumbers.size() != 0) {
@@ -189,7 +191,7 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
try {
moveToNextPipeLog(pipeData.getSerialNumber());
} catch (IOException e) {
- logger.warn(String.format("Move to next pipe log %s error, because %s.", pipeData, e));
+ logger.error(String.format("Move to next pipe log %s error, because %s.", pipeData, e));
}
}
if (!inputDeque.offer(pipeData)) {
@@ -213,7 +215,7 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
outputStream.close();
}
File newPipeLog = new File(pipeLogDir, SyncConstant.getPipeLogName(startSerialNumber));
- createFile(newPipeLog);
+ SyncPathUtil.createFile(newPipeLog);
outputStream = new DataOutputStream(new FileOutputStream(newPipeLog));
pipeLogStartNumber.offer(startSerialNumber);
@@ -238,7 +240,7 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
if (!outputDeque.isEmpty()) {
return outputDeque.poll();
} else if (outputDeque != inputDeque) {
- if (pipeLogStartNumber.isEmpty()) {
+ if (pipeLogStartNumber.isEmpty() || lastSerialNumber == Long.MIN_VALUE) {
return null;
}
@@ -293,11 +295,9 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
PipeData pipeData = null;
try {
synchronized (waitLock) {
- pipeData = pullOnePipeData(commitSerialNumber);
- if (pipeData == null) {
+ while ((pipeData = pullOnePipeData(commitSerialNumber)) == null) {
waitLock.wait();
waitLock.notifyAll();
- pipeData = pullOnePipeData(commitSerialNumber);
}
}
} catch (IOException e) {
@@ -312,18 +312,25 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
@Override
public void commit() {
- deletePipeData();
+ if (pullSerialNumber == Long.MIN_VALUE) {
+ return;
+ }
+ commit(pullSerialNumber);
+ }
+
+ public void commit(long serialNumber) {
+ deletePipeData(serialNumber);
deletePipeLog();
serializeCommitSerialNumber();
}
- private void deletePipeData() {
- while (commitSerialNumber < pullSerialNumber) {
+ private void deletePipeData(long serialNumber) {
+ while (commitSerialNumber < serialNumber) {
commitSerialNumber += 1;
try {
PipeData commitData = pullOnePipeData(commitSerialNumber);
if (PipeData.Type.TSFILE.equals(commitData.getType())) {
- List<File> tsFiles = ((TsFilePipeData) commitData).getTsFiles();
+ List<File> tsFiles = ((TsFilePipeData) commitData).getTsFiles(false);
for (File file : tsFiles) {
Files.deleteIfExists(file.toPath());
}
@@ -381,6 +388,16 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
/** common */
@Override
+ public synchronized boolean isEmpty() {
+ if (outputDeque == null || pipeLogStartNumber.isEmpty()) {
+ return true;
+ }
+ return pipeLogStartNumber.size() == 1
+ && outputDeque.isEmpty()
+ && (inputDeque == null || inputDeque.isEmpty());
+ }
+
+ @Override
public void clear() {
try {
if (outputStream != null) {
@@ -404,13 +421,6 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
}
}
- private boolean createFile(File file) throws IOException {
- if (!file.getParentFile().exists()) {
- file.getParentFile().mkdirs();
- }
- return file.createNewFile();
- }
-
public static List<PipeData> parsePipeLog(File file) throws IOException {
List<PipeData> pipeData = new ArrayList<>();
try (DataInputStream inputStream = new DataInputStream(new FileInputStream(file))) {
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/PipeDataQueue.java b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/PipeDataQueue.java
index 6adac03..159aaf7 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/PipeDataQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/PipeDataQueue.java
@@ -31,5 +31,7 @@ public interface PipeDataQueue {
void commit();
+ boolean isEmpty();
+
void clear();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
index 983f17c..bd1074c 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
@@ -33,6 +33,7 @@ import org.apache.iotdb.db.qp.utils.DatetimeUtils;
import org.apache.iotdb.db.query.dataset.ListDataSet;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.service.transport.thrift.ResponseType;
import org.apache.iotdb.service.transport.thrift.SyncRequest;
import org.apache.iotdb.service.transport.thrift.SyncResponse;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -96,28 +97,33 @@ public class ReceiverService implements IService {
/** heartbeat RPC handle */
// TODO: define exception
// TODO: this is a mock interface
- public SyncResponse recMsg(SyncRequest request) throws IOException {
- switch (request.getType()) {
- case HEARTBEAT:
- List<PipeMessage> messages =
- receiverManager.getPipeMessages(
- request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
- break;
- case CREATE:
- createPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
- break;
- case START:
- startPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
- break;
- case STOP:
- stopPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
- break;
- case DROP:
- dropPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
- break;
+ public SyncResponse recMsg(SyncRequest request) {
+ try {
+ switch (request.getType()) {
+ case HEARTBEAT:
+ List<PipeMessage> messages =
+ receiverManager.getPipeMessages(
+ request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
+ break;
+ case CREATE:
+ createPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
+ break;
+ case START:
+ startPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
+ break;
+ case STOP:
+ stopPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
+ break;
+ case DROP:
+ dropPipe(request.getPipeName(), request.getRemoteIp(), request.getCreateTime());
+ break;
+ }
+ // TODO: complete implement
+ } catch (IOException e) {
+ // TODO: complete
+ logger.error("Need to complete.");
}
- // TODO: complete implement
- return null;
+ return new SyncResponse(ResponseType.INFO, "");
}
/** create and start a new pipe named pipeName */
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/IoTDBPipeSink.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/IoTDBPipeSink.java
index 5180919..4981cae 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/IoTDBPipeSink.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/IoTDBPipeSink.java
@@ -55,6 +55,14 @@ public class IoTDBPipeSink implements PipeSink {
}
}
+ public String getIp() {
+ return ip;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
@Override
public String getName() {
return name;
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/Pipe.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/Pipe.java
index 5793af8..ec0df18 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/Pipe.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/Pipe.java
@@ -20,13 +20,14 @@
package org.apache.iotdb.db.newsync.sender.pipe;
import org.apache.iotdb.db.exception.PipeException;
+import org.apache.iotdb.db.newsync.pipedata.PipeData;
public interface Pipe {
void start() throws PipeException;
void stop() throws PipeException;
- void drop();
+ void drop() throws PipeException;
String getName();
@@ -36,6 +37,10 @@ public interface Pipe {
PipeStatus getStatus();
+ PipeData take() throws InterruptedException;
+
+ void commit();
+
// a new pipe should be stop status
enum PipeStatus {
RUNNING,
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/TransportHandler.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/TransportHandler.java
new file mode 100644
index 0000000..138458a
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/TransportHandler.java
@@ -0,0 +1,136 @@
+/*
+ * 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.iotdb.db.newsync.sender.pipe;
+
+import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.db.concurrent.ThreadName;
+import org.apache.iotdb.db.exception.SyncConnectionException;
+import org.apache.iotdb.db.newsync.conf.SyncConstant;
+import org.apache.iotdb.db.newsync.sender.service.SenderService;
+import org.apache.iotdb.db.newsync.transport.client.ITransportClient;
+import org.apache.iotdb.service.transport.thrift.RequestType;
+import org.apache.iotdb.service.transport.thrift.SyncRequest;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class TransportHandler {
+ private static final Logger logger = LoggerFactory.getLogger(TransportHandler.class);
+
+ private final String pipeName;
+ private final String localIp;
+ private final long createTime;
+ private final ITransportClient transportClient;
+
+ private final ExecutorService transportExecutorService;
+ private Future transportFuture;
+
+ private final ScheduledExecutorService heartbeatExecutorService;
+ private Future heartbeatFuture;
+
+ public TransportHandler(ITransportClient transportClient, String pipeName, long createTime)
+ throws SyncConnectionException {
+ this.pipeName = pipeName;
+ this.createTime = createTime;
+ this.transportClient = transportClient;
+
+ this.transportExecutorService =
+ IoTDBThreadPoolFactory.newSingleThreadExecutor(
+ ThreadName.SYNC_SENDER_PIPE.getName() + "-" + pipeName);
+ this.heartbeatExecutorService =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.SYNC_SENDER_HEARTBEAT.getName() + "-" + pipeName);
+
+ String localIp1;
+ try {
+ localIp1 = InetAddress.getLocalHost().getHostAddress();
+ } catch (UnknownHostException e) {
+ logger.error(
+ String.format("Get local host error when create transport handler, because %s.", e));
+ localIp1 = SyncConstant.UNKNOWN_IP;
+ }
+ this.localIp = localIp1;
+
+ create();
+ }
+
+ private void create() throws SyncConnectionException {
+ transportClient.heartbeat(new SyncRequest(RequestType.CREATE, pipeName, localIp, createTime));
+ }
+
+ public void start() throws SyncConnectionException {
+ transportClient.heartbeat(new SyncRequest(RequestType.START, pipeName, localIp, createTime));
+
+ transportFuture = transportExecutorService.submit(transportClient);
+ heartbeatFuture =
+ heartbeatExecutorService.scheduleWithFixedDelay(
+ this::sendHeartbeat,
+ SyncConstant.DEFAULT_HEARTBEAT_DELAY_SECONDS,
+ SyncConstant.DEFAULT_HEARTBEAT_DELAY_SECONDS,
+ TimeUnit.SECONDS);
+ }
+
+ public void stop() throws SyncConnectionException {
+ transportClient.heartbeat(new SyncRequest(RequestType.STOP, pipeName, localIp, createTime));
+ if (transportFuture != null) {
+ transportFuture.cancel(true);
+ }
+ if (heartbeatFuture != null) {
+ heartbeatFuture.cancel(true);
+ }
+ }
+
+ public boolean close() throws InterruptedException, SyncConnectionException {
+ boolean isClosed;
+ transportExecutorService.shutdownNow();
+ isClosed =
+ transportExecutorService.awaitTermination(
+ SyncConstant.DEFAULT_WAITTING_FOR_STOP_MILLISECONDS, TimeUnit.MILLISECONDS);
+ heartbeatExecutorService.shutdownNow();
+ isClosed &=
+ heartbeatExecutorService.awaitTermination(
+ SyncConstant.DEFAULT_WAITTING_FOR_STOP_MILLISECONDS, TimeUnit.MILLISECONDS);
+ if (isClosed) {
+ transportClient.heartbeat(new SyncRequest(RequestType.DROP, pipeName, localIp, createTime));
+ }
+ return isClosed;
+ }
+
+ private void sendHeartbeat() {
+ try {
+ SenderService.getInstance()
+ .recMsg(
+ transportClient.heartbeat(
+ new SyncRequest(RequestType.HEARTBEAT, pipeName, localIp, createTime)));
+ } catch (SyncConnectionException e) {
+ logger.warn(
+ String.format(
+ "Pipe %s sends heartbeat to receiver error, skip this time, because %s.",
+ pipeName, e));
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/TsFilePipe.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/TsFilePipe.java
index 6fd1fb9..d597874 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/TsFilePipe.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/pipe/TsFilePipe.java
@@ -19,12 +19,11 @@
*/
package org.apache.iotdb.db.newsync.sender.pipe;
-import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.storagegroup.virtualSg.StorageGroupManager;
import org.apache.iotdb.db.exception.PipeException;
+import org.apache.iotdb.db.exception.SyncConnectionException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.newsync.conf.SyncConstant;
@@ -33,7 +32,7 @@ import org.apache.iotdb.db.newsync.pipedata.DeletionPipeData;
import org.apache.iotdb.db.newsync.pipedata.PipeData;
import org.apache.iotdb.db.newsync.pipedata.SchemaPipeData;
import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
-import org.apache.iotdb.db.newsync.sender.recovery.TsFilePipeLogAnalyzer;
+import org.apache.iotdb.db.newsync.pipedata.queue.BufferedPipeDataQueue;
import org.apache.iotdb.db.newsync.sender.recovery.TsFilePipeLogger;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
@@ -49,8 +48,7 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.BlockingDeque;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import static org.apache.iotdb.db.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
@@ -64,15 +62,17 @@ public class TsFilePipe implements Pipe {
private final long dataStartTime;
private final boolean syncDelOp;
- private final ExecutorService singleExecutorService;
+ private final BufferedPipeDataQueue historyQueue;
+ private final BufferedPipeDataQueue realTimeQueue;
private final TsFilePipeLogger pipeLog;
private final ReentrantLock collectRealTimeDataLock;
- private BlockingDeque<PipeData> pipeDataDeque;
+ private boolean isCollectingRealTimeData;
private long maxSerialNumber;
+ private TransportHandler transportHandler;
+
private PipeStatus status;
- private boolean isCollectingRealTimeData;
public TsFilePipe(
long createTime, String name, PipeSink pipeSink, long dataStartTime, boolean syncDelOp) {
@@ -82,25 +82,21 @@ public class TsFilePipe implements Pipe {
this.dataStartTime = dataStartTime;
this.syncDelOp = syncDelOp;
+ this.historyQueue =
+ new BufferedPipeDataQueue(SyncPathUtil.getSenderHistoryPipeDataDir(name, createTime));
+ this.realTimeQueue =
+ new BufferedPipeDataQueue(SyncPathUtil.getSenderRealTimePipeDataDir(name, createTime));
this.pipeLog = new TsFilePipeLogger(this);
- this.singleExecutorService =
- IoTDBThreadPoolFactory.newSingleThreadExecutor(
- ThreadName.PIPE_SERVICE.getName() + "-" + name);
-
- recover();
-
this.collectRealTimeDataLock = new ReentrantLock();
- this.status = PipeStatus.STOP;
this.isCollectingRealTimeData = false;
+ this.maxSerialNumber = Math.max(0L, realTimeQueue.getLastMaxSerialNumber());
+
+ this.status = PipeStatus.STOP;
}
- private void recover() {
- this.pipeDataDeque = new TsFilePipeLogAnalyzer(this).recover();
- this.maxSerialNumber = 0L;
- if (pipeDataDeque.size() != 0) {
- this.maxSerialNumber = Math.max(maxSerialNumber, pipeDataDeque.getLast().getSerialNumber());
- }
+ public void setTransportHandler(TransportHandler transportHandler) {
+ this.transportHandler = transportHandler;
}
@Override
@@ -113,7 +109,7 @@ public class TsFilePipe implements Pipe {
}
try {
- if (!new TsFilePipeLogAnalyzer(this).isCollectFinished()) {
+ if (!pipeLog.isCollectFinished()) {
pipeLog.clear();
collectData();
pipeLog.finishCollect();
@@ -124,8 +120,7 @@ public class TsFilePipe implements Pipe {
isCollectingRealTimeData = true;
}
- singleExecutorService.submit(this::transport);
- //
+ transportHandler.start();
status = PipeStatus.RUNNING;
} catch (IOException e) {
logger.error(
@@ -133,6 +128,12 @@ public class TsFilePipe implements Pipe {
"Clear pipe dir %s error, because %s.",
SyncPathUtil.getSenderPipeDir(name, createTime), e));
throw new PipeException("Start error, can not clear pipe log.");
+ } catch (SyncConnectionException e) {
+ logger.warn(String.format("%s pipe sends START request error, because %s.", name, e));
+ throw new PipeException(
+ String.format(
+ "Start error, because %s, please check receiver server is open and net.",
+ e.getMessage()));
}
}
@@ -155,8 +156,9 @@ public class TsFilePipe implements Pipe {
long serialNumber = 1 - historyTsFilesSize + i;
try {
File hardLink =
- pipeLog.addHistoryTsFile(historyTsFiles.get(i).left, historyTsFiles.get(i).right);
- historyData.add(new TsFilePipeData(hardLink.getPath(), serialNumber));
+ pipeLog.createTsFileAndModsHardlink(
+ historyTsFiles.get(i).left, historyTsFiles.get(i).right);
+ historyData.add(new TsFilePipeData(hardLink.getParent(), hardLink.getName(), serialNumber));
} catch (IOException e) {
logger.warn(
String.format(
@@ -168,19 +170,7 @@ public class TsFilePipe implements Pipe {
// add history data into blocking deque
int historyDataSize = historyData.size();
for (int i = 0; i < historyDataSize; i++) {
- pipeDataDeque.addFirst(historyData.get(historyDataSize - 1 - i));
- }
- // record history data
- for (int i = 0; i < historyDataSize; i++) {
- PipeData data = historyData.get(i);
- try {
- pipeLog.addHistoryPipeData(data);
- } catch (IOException e) {
- logger.warn(
- String.format(
- "Record history pipe data %s on disk error, serial number is %d, because %s.",
- data, data.getSerialNumber(), e));
- }
+ historyQueue.offer(historyData.get(i));
}
}
@@ -215,13 +205,7 @@ public class TsFilePipe implements Pipe {
try {
maxSerialNumber += 1L;
PipeData metaData = new SchemaPipeData(plan, maxSerialNumber);
- collectRealTimePipeData(metaData); // ensure can be transport
- pipeLog.addRealTimePipeData(metaData);
- } catch (IOException e) {
- logger.warn(
- String.format(
- "Record plan %s on disk error, serial number is %d, because %s.",
- plan, maxSerialNumber, e));
+ realTimeQueue.offer(metaData);
} finally {
collectRealTimeDataLock.unlock();
}
@@ -273,16 +257,10 @@ public class TsFilePipe implements Pipe {
deletion.getEndTime());
maxSerialNumber += 1L;
PipeData deletionData = new DeletionPipeData(splitDeletion, maxSerialNumber);
- pipeLog.addRealTimePipeData(deletionData);
- collectRealTimePipeData(deletionData);
+ realTimeQueue.offer(deletionData);
}
} catch (MetadataException e) {
logger.warn(String.format("Collect deletion %s error, because %s.", deletion, e));
- } catch (IOException e) {
- logger.warn(
- String.format(
- "Record deletion %s on disk error, serial number is %d, because %s.",
- deletion, maxSerialNumber, e));
} finally {
collectRealTimeDataLock.unlock();
}
@@ -292,14 +270,14 @@ public class TsFilePipe implements Pipe {
collectRealTimeDataLock.lock();
try {
maxSerialNumber += 1L;
+ File hardlink = pipeLog.createTsFileHardlink(tsFile);
PipeData tsFileData =
- new TsFilePipeData(pipeLog.addRealTimeTsFile(tsFile).getPath(), maxSerialNumber);
- pipeLog.addRealTimePipeData(tsFileData);
- collectRealTimePipeData(tsFileData);
+ new TsFilePipeData(hardlink.getParent(), hardlink.getName(), maxSerialNumber);
+ realTimeQueue.offer(tsFileData);
} catch (IOException e) {
logger.warn(
String.format(
- "Record tsfile %s on disk error, serial number is %d, because %s.",
+ "Create Hardlink tsfile %s on disk error, serial number is %d, because %s.",
tsFile.getPath(), maxSerialNumber, e));
} finally {
collectRealTimeDataLock.unlock();
@@ -308,7 +286,7 @@ public class TsFilePipe implements Pipe {
public void collectRealTimeTsFileResource(File tsFile) {
try {
- pipeLog.addTsFileResource(tsFile);
+ pipeLog.createTsFileResourceHardlink(tsFile);
} catch (IOException e) {
logger.warn(
String.format(
@@ -316,84 +294,40 @@ public class TsFilePipe implements Pipe {
}
}
- private void collectRealTimePipeData(PipeData data) {
- pipeDataDeque.offer(data);
- synchronized (pipeDataDeque) {
- pipeDataDeque.notifyAll();
- }
- }
-
/** transport data * */
- private void transport() {
- // handshake
- try {
- while (true) {
- if (status == PipeStatus.STOP || status == PipeStatus.DROP) {
- logger.info(String.format("TsFile pipe %s stops transporting data by command.", name));
- break;
- }
-
- PipeData data;
- try {
- synchronized (pipeDataDeque) {
- if (pipeDataDeque.isEmpty()) {
- pipeDataDeque.wait();
- pipeDataDeque.notifyAll();
- }
- data = pipeDataDeque.poll();
- }
- } catch (InterruptedException e) {
- logger.warn(String.format("TsFile pipe %s has been interrupted.", name));
- continue;
- }
-
- if (data == null) {
- continue;
- }
- // data.sendToTransport();
- // pipeLog.removePipeData(data.getSerialNumber);
- // data.sendToTransport();
- // Thread.sleep(1000);
- // pipeLog.removePipeData(data.getSerialNumber());
- }
- } catch (Exception e) {
- logger.error(String.format("TsFile pipe %s stops transportng data, because %s.", name, e));
+ @Override
+ public PipeData take() throws InterruptedException {
+ if (!historyQueue.isEmpty()) {
+ return historyQueue.take();
}
+ return realTimeQueue.take();
}
public List<PipeData> pull(long serialNumber) {
- if (pipeDataDeque.isEmpty()) {
- return null;
- }
List<PipeData> pullPipeData = new ArrayList<>();
- PipeData data = pipeDataDeque.poll();
- while (data.getSerialNumber() <= serialNumber) {
- pullPipeData.add(data);
- if (pipeDataDeque.isEmpty()) {
- break;
- } else {
- data = pipeDataDeque.poll();
- }
+ if (!historyQueue.isEmpty()) {
+ pullPipeData.addAll(historyQueue.pull(serialNumber));
}
-
- int pullPipeDataSize = pullPipeData.size();
- for (int i = 0; i < pullPipeDataSize; i++) {
- pipeDataDeque.addFirst(pullPipeData.get(pullPipeDataSize - i - 1));
+ if (serialNumber > 0) {
+ pullPipeData.addAll(realTimeQueue.pull(serialNumber));
}
return pullPipeData;
}
+ @Override
+ public void commit() {
+ if (!historyQueue.isEmpty()) {
+ historyQueue.commit();
+ }
+ realTimeQueue.commit();
+ }
+
public void commit(long serialNumber) {
- while (!pipeDataDeque.isEmpty() && pipeDataDeque.peek().getSerialNumber() <= serialNumber) {
- PipeData data = pipeDataDeque.poll();
- try {
- pipeLog.removePipeData(data);
- } catch (IOException e) {
- logger.warn(
- String.format(
- "Commit pipe data %s error, serial number is %s, because %s",
- data, data.getSerialNumber(), e));
- }
+ if (!historyQueue.isEmpty()) {
+ historyQueue.commit(serialNumber);
+ }
+ if (serialNumber > 0) {
+ realTimeQueue.commit(serialNumber);
}
}
@@ -409,47 +343,60 @@ public class TsFilePipe implements Pipe {
registerTsFile();
isCollectingRealTimeData = true;
}
- status = PipeStatus.STOP;
- synchronized (pipeDataDeque) {
- pipeDataDeque.notifyAll();
+
+ try {
+ transportHandler.stop();
+ } catch (SyncConnectionException e) {
+ logger.warn(String.format("%s pipe sends STOP request error, because %s.", name, e));
+ throw new PipeException(
+ "Stop error, can not connect to receiver, please check receiver server is open and net.");
}
+ status = PipeStatus.STOP;
}
@Override
- public synchronized void drop() {
+ public synchronized void drop() throws PipeException {
if (status == PipeStatus.DROP) {
return;
}
- status = PipeStatus.DROP;
- synchronized (pipeDataDeque) {
- pipeDataDeque.notifyAll();
- }
clear();
+ status = PipeStatus.DROP;
}
- private void clear() {
+ private void clear() throws PipeException {
deregisterMetadata();
deregisterTsFile();
-
- singleExecutorService.shutdown();
+ isCollectingRealTimeData = false;
try {
- Thread.sleep(SyncConstant.DEFAULT_WAITING_FOR_DEREGISTER_MILLISECONDS);
+ if (!transportHandler.close()) {
+ throw new PipeException(
+ String.format(
+ "Close pipe %s transport when waiting for stop pipe after %s %s.",
+ name,
+ SyncConstant.DEFAULT_WAITTING_FOR_STOP_MILLISECONDS,
+ TimeUnit.MILLISECONDS.name()));
+ }
} catch (InterruptedException e) {
logger.warn(
String.format(
- "Be interrupted when pipe %s %d waiting for deregister from tsfile, because %s.",
- name, createTime, e));
+ "Interrupted when waiting for clear pipe %s %d, because %s", name, createTime, e));
+ throw new PipeException("Drop error, be interrupted, please try again.");
+ } catch (SyncConnectionException e) {
+ logger.warn(String.format("%s pipe sends START request error, because %s.", name, e));
+ throw new PipeException(
+ "Drop error, can not connect to receiver, please check receiver server is open and net, and try again.");
+ // the status is running, but transport client is stop.
}
+
try {
+ historyQueue.clear();
+ realTimeQueue.clear();
pipeLog.clear();
} catch (IOException e) {
logger.warn(String.format("Clear pipe %s %d error, because %s.", name, createTime, e));
}
-
- pipeDataDeque = null;
- isCollectingRealTimeData = false;
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogAnalyzer.java
index 2504145..e4ed459 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogAnalyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogAnalyzer.java
@@ -47,6 +47,7 @@ public class SenderLogAnalyzer {
private Pipe runningPipe;
private Pipe.PipeStatus runningPipeStatus;
+ private String runningMsg;
public SenderLogAnalyzer() throws IOException {
senderLog = new File(SyncPathUtil.getSysDir(), SyncConstant.SENDER_LOG_NAME);
@@ -54,8 +55,9 @@ public class SenderLogAnalyzer {
senderLog.createNewFile();
}
- pipeSinks = new HashMap<>();
- pipes = new ArrayList<>();
+ this.pipeSinks = new HashMap<>();
+ this.pipes = new ArrayList<>();
+ this.runningMsg = "";
}
public void recover() throws IOException {
@@ -96,12 +98,15 @@ public class SenderLogAnalyzer {
pipes.add(runningPipe);
runningPipeStatus = runningPipe.getStatus();
runningPipe.stop();
+ runningMsg = "";
break;
case STOP_PIPE: // ignore status check
runningPipeStatus = Pipe.PipeStatus.STOP;
+ appendMsg(parseStrings);
break;
case START_PIPE:
runningPipeStatus = Pipe.PipeStatus.RUNNING;
+ appendMsg(parseStrings);
break;
case DROP_PIPE:
runningPipeStatus = Pipe.PipeStatus.DROP;
@@ -141,6 +146,15 @@ public class SenderLogAnalyzer {
br.close();
}
+ private void appendMsg(String[] parseStrings) {
+ if (parseStrings.length == 3) {
+ if (runningMsg.length() > 0) {
+ runningMsg += System.lineSeparator();
+ }
+ runningMsg += parseStrings[2];
+ }
+ }
+
public Map<String, PipeSink> getRecoveryAllPipeSinks() {
return pipeSinks;
}
@@ -152,4 +166,8 @@ public class SenderLogAnalyzer {
public Pipe getRecoveryRunningPipe() {
return runningPipe;
}
+
+ public String getRecoveryRunningMsg() {
+ return runningMsg;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogger.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogger.java
index 9afba24..9518d5c 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogger.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/SenderLogger.java
@@ -114,6 +114,21 @@ public class SenderLogger {
}
}
+ public void recordMsg(String pipeName, Operator.OperatorType type, String msg) {
+ getBufferedWriter();
+ try {
+ bw.write(type.name());
+ bw.write(SyncConstant.SENDER_LOG_SPLIT_CHARACTER);
+ bw.write(pipeName);
+ bw.write(SyncConstant.SENDER_LOG_SPLIT_CHARACTER);
+ bw.write(msg);
+ bw.newLine();
+ bw.flush();
+ } catch (IOException e) {
+ logger.warn(String.format("Record msg %s error, because %s.", msg, e));
+ }
+ }
+
public void close() {
try {
if (bw != null) {
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/TsFilePipeLogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/TsFilePipeLogAnalyzer.java
deleted file mode 100644
index f0ac0ef..0000000
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/TsFilePipeLogAnalyzer.java
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- * 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.iotdb.db.newsync.sender.recovery;
-
-import org.apache.iotdb.db.exception.metadata.IllegalPathException;
-import org.apache.iotdb.db.newsync.conf.SyncConstant;
-import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
-import org.apache.iotdb.db.newsync.pipedata.PipeData;
-import org.apache.iotdb.db.newsync.sender.pipe.TsFilePipe;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.BufferedReader;
-import java.io.DataInputStream;
-import java.io.EOFException;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileReader;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.BlockingDeque;
-import java.util.concurrent.LinkedBlockingDeque;
-
-public class TsFilePipeLogAnalyzer {
- private static final Logger logger = LoggerFactory.getLogger(TsFilePipeLogAnalyzer.class);
-
- private final String pipeDir;
- private final String pipeLogDir;
-
- private BlockingDeque<PipeData> pipeDataDeque;
- private long removeSerialNumber;
-
- public TsFilePipeLogAnalyzer(TsFilePipe pipe) {
- pipeDir = SyncPathUtil.getSenderPipeDir(pipe.getName(), pipe.getCreateTime());
- pipeLogDir = new File(pipeDir, SyncConstant.PIPE_LOG_DIR_NAME).getPath();
- }
-
- public BlockingDeque<PipeData> recover() {
- pipeDataDeque = new LinkedBlockingDeque<>();
- removeSerialNumber = Long.MIN_VALUE;
-
- if (!new File(pipeDir).exists()) {
- return pipeDataDeque;
- }
-
- deserializeRemoveSerialNumber();
- recoverHistoryData();
- recoverRealTimeData();
-
- return pipeDataDeque;
- }
-
- private void deserializeRemoveSerialNumber() {
- File removeSerialNumberLog = new File(pipeLogDir, SyncConstant.COMMIT_LOG_NAME);
- if (!removeSerialNumberLog.exists()) {
- return;
- }
-
- try {
- BufferedReader br = new BufferedReader(new FileReader(removeSerialNumberLog));
- long lineIndex = 0;
- String readLine;
- while ((readLine = br.readLine()) != null) {
- lineIndex += 1;
- try {
- removeSerialNumber = Long.parseLong(readLine);
- } catch (NumberFormatException e) {
- logger.warn(String.format("Can not parse Long in line %s, skip it.", lineIndex));
- }
- }
- br.close();
- } catch (IOException e) {
- logger.warn(
- String.format(
- "deserialize remove serial number error, remove serial number has been set to %d, because %s",
- removeSerialNumber, e));
- }
- }
-
- private void recoverHistoryData() {
- File historyPipeLog = new File(pipeLogDir, SyncConstant.HISTORY_PIPE_LOG_NAME);
- if (!historyPipeLog.exists()) {
- return;
- }
-
- if (removeSerialNumber < 0) {
- try {
- List<PipeData> historyPipeData = parseFile(historyPipeLog);
- for (PipeData data : historyPipeData)
- if (data.getSerialNumber() > removeSerialNumber) {
- pipeDataDeque.offer(data);
- }
- } catch (IOException e) {
- logger.error(
- String.format(
- "Can not parse history pipe log %s, because %s", historyPipeLog.getPath(), e));
- }
- } else {
- try {
- Files.delete(historyPipeLog.toPath());
- } catch (IOException e) {
- logger.warn(
- String.format(
- "Can not delete history pipe log %s, because %s", historyPipeLog.getPath(), e));
- }
- }
- }
-
- private void recoverRealTimeData() {
- File pipeLogDir = new File(this.pipeLogDir);
- if (!pipeLogDir.exists()) {
- return;
- }
-
- List<Long> startNumbers = new ArrayList<>();
- for (File file : pipeLogDir.listFiles())
- if (file.getName().endsWith(SyncConstant.PIPE_LOG_NAME_SUFFIX)) {
- startNumbers.add(SyncConstant.getSerialNumberFromPipeLogName(file.getName()));
- }
- if (startNumbers.size() != 0) {
- Collections.sort(startNumbers);
- for (Long startNumber : startNumbers) {
- File realTimePipeLog = new File(this.pipeLogDir, SyncConstant.getPipeLogName(startNumber));
- try {
- List<PipeData> realTimeData = parseFile(realTimePipeLog);
- for (PipeData data : realTimeData)
- if (data.getSerialNumber() > removeSerialNumber) {
- pipeDataDeque.offer(data);
- }
- } catch (IOException e) {
- logger.error(
- String.format(
- "Can not parse real time pipe log %s, because %s", realTimePipeLog.getPath(), e));
- }
- }
- }
- }
-
- public boolean isCollectFinished() {
- return new File(pipeDir, SyncConstant.FINISH_COLLECT_LOCK_NAME).exists();
- }
-
- public static List<PipeData> parseFile(File file) throws IOException {
- List<PipeData> pipeData = new ArrayList<>();
- try (DataInputStream inputStream = new DataInputStream(new FileInputStream(file))) {
- while (true) {
- pipeData.add(PipeData.deserialize(inputStream));
- }
- } catch (EOFException e) {
- logger.info(String.format("Finish parsing pipeLog %s.", file.getPath()));
- } catch (IllegalPathException e) {
- logger.error(String.format("Parsing pipeLog %s error, because %s", file.getPath(), e));
- throw new IOException(e);
- }
-
- return pipeData;
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/TsFilePipeLogger.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/TsFilePipeLogger.java
index 4169e8a..5a2e3da 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/TsFilePipeLogger.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/recovery/TsFilePipeLogger.java
@@ -24,8 +24,6 @@ import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.newsync.conf.SyncConstant;
import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
-import org.apache.iotdb.db.newsync.pipedata.PipeData;
-import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
import org.apache.iotdb.db.newsync.sender.pipe.TsFilePipe;
import org.apache.iotdb.db.utils.FileUtils;
@@ -33,45 +31,26 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedWriter;
-import java.io.DataOutputStream;
import java.io.File;
-import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
-import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.BlockingDeque;
-import java.util.concurrent.LinkedBlockingDeque;
public class TsFilePipeLogger {
private static final Logger logger = LoggerFactory.getLogger(TsFilePipeLogger.class);
private final String pipeDir;
private final String tsFileDir;
- private final String pipeLogDir;
-
- private DataOutputStream historyOutputStream;
-
- private BlockingDeque<Long> realTimePipeLogStartNumber;
- private DataOutputStream realTimeOutputStream;
- private long currentPipeLogSize;
-
- private BufferedWriter removeSerialNumberWriter;
- private long currentRemoveLogSize;
public TsFilePipeLogger(TsFilePipe tsFilePipe) {
pipeDir = SyncPathUtil.getSenderPipeDir(tsFilePipe.getName(), tsFilePipe.getCreateTime());
tsFileDir = new File(pipeDir, SyncConstant.FILE_DATA_DIR_NAME).getPath();
- pipeLogDir = new File(pipeDir, SyncConstant.PIPE_LOG_DIR_NAME).getPath();
}
/** make hard link for tsfile * */
- public File addHistoryTsFile(File tsFile, long modsOffset) throws IOException {
+ public File createTsFileAndModsHardlink(File tsFile, long modsOffset) throws IOException {
File mods = new File(tsFile.getPath() + ModificationFile.FILE_SUFFIX);
if (mods.exists()) {
@@ -86,17 +65,17 @@ public class TsFilePipeLogger {
"Can not find %s mods to create hard link. The mods offset is %d.",
mods.getPath(), modsOffset));
}
- addTsFileResource(tsFile);
- return addRealTimeTsFile(tsFile);
+ createTsFileResourceHardlink(tsFile);
+ return createTsFileHardlink(tsFile);
}
private void serializeModsOffset(File modsOffsetFile, long modsOffset) {
try {
- createFile(modsOffsetFile);
- BufferedWriter bw = new BufferedWriter(new FileWriter(modsOffsetFile));
- bw.write(String.valueOf(modsOffset));
- bw.flush();
- bw.close();
+ SyncPathUtil.createFile(modsOffsetFile);
+ try (BufferedWriter bw = new BufferedWriter(new FileWriter(modsOffsetFile))) {
+ bw.write(String.valueOf(modsOffset));
+ bw.flush();
+ }
} catch (IOException e) {
logger.warn(
String.format(
@@ -105,11 +84,11 @@ public class TsFilePipeLogger {
}
}
- public File addRealTimeTsFile(File tsFile) throws IOException {
+ public File createTsFileHardlink(File tsFile) throws IOException {
return createHardLink(tsFile);
}
- public void addTsFileResource(File tsFile) throws IOException {
+ public void createTsFileResourceHardlink(File tsFile) throws IOException {
File tsFileResource = new File(tsFile.getPath() + TsFileResource.RESOURCE_SUFFIX);
try {
createHardLink(tsFileResource);
@@ -118,7 +97,8 @@ public class TsFilePipeLogger {
String.format(
"Record tsfile resource %s on disk error, make a empty to close it.",
tsFileResource.getPath()));
- createFile(new File(tsFileDir, getRelativeFilePath(tsFileResource)));
+ SyncPathUtil.createFile(
+ new File(tsFileDir, getRelativeFilePath(tsFileResource))); // create an empty resource
}
}
@@ -139,167 +119,17 @@ public class TsFilePipeLogger {
while (!file.getName().equals(IoTDBConstant.SEQUENCE_FLODER_NAME)
&& !file.getName().equals(IoTDBConstant.UNSEQUENCE_FLODER_NAME)) {
file = file.getParentFile();
- builder = new StringBuilder(file.getName()).append(File.separator).append(builder);
+ builder =
+ new StringBuilder(file.getName())
+ .append(IoTDBConstant.FILE_NAME_SEPARATOR)
+ .append(builder);
}
return builder.toString();
}
- /** add pipe log data */
- public void addHistoryPipeData(PipeData pipeData) throws IOException {
- getHistoryOutputStream();
- pipeData.serialize(historyOutputStream);
- historyOutputStream.flush();
- }
-
- private void getHistoryOutputStream() throws IOException {
- if (historyOutputStream != null) {
- return;
- }
-
- // recover history pipe log
- File logDir = new File(pipeLogDir);
- logDir.mkdirs();
- File historyPipeLog = new File(pipeLogDir, SyncConstant.HISTORY_PIPE_LOG_NAME);
- createFile(historyPipeLog);
- historyOutputStream = new DataOutputStream(new FileOutputStream(historyPipeLog, true));
- }
-
- public synchronized void addRealTimePipeData(PipeData pipeData) throws IOException {
- getRealTimeOutputStream(pipeData.getSerialNumber());
- currentPipeLogSize += pipeData.serialize(realTimeOutputStream);
- realTimeOutputStream.flush();
- }
-
- private void getRealTimeOutputStream(long serialNumber) throws IOException {
- if (realTimeOutputStream == null) {
- // recover real time pipe log
- if (realTimePipeLogStartNumber == null) {
- recoverRealTimePipeLogStartNumber();
- }
- if (!realTimePipeLogStartNumber.isEmpty()) {
- File writingPipeLog =
- new File(
- pipeLogDir, SyncConstant.getPipeLogName(realTimePipeLogStartNumber.peekLast()));
- realTimeOutputStream = new DataOutputStream(new FileOutputStream(writingPipeLog, true));
- currentPipeLogSize = writingPipeLog.length();
- } else {
- moveToNextPipeLog(serialNumber);
- }
- }
-
- if (currentPipeLogSize > SyncConstant.DEFAULT_PIPE_LOG_SIZE_IN_BYTE) {
- moveToNextPipeLog(serialNumber);
- }
- }
-
- private void recoverRealTimePipeLogStartNumber() {
- realTimePipeLogStartNumber = new LinkedBlockingDeque<>();
- File logDir = new File(pipeLogDir);
- List<Long> startNumbers = new ArrayList<>();
-
- logDir.mkdirs();
- for (File file : logDir.listFiles())
- if (file.getName().endsWith(SyncConstant.PIPE_LOG_NAME_SUFFIX)) {
- startNumbers.add(SyncConstant.getSerialNumberFromPipeLogName(file.getName()));
- }
- if (startNumbers.size() != 0) {
- Collections.sort(startNumbers);
- for (Long startTime : startNumbers) {
- realTimePipeLogStartNumber.offer(startTime);
- }
- }
- }
-
- private void moveToNextPipeLog(long startSerialNumber) throws IOException {
- if (realTimeOutputStream != null) {
- realTimeOutputStream.close();
- }
- File newPipeLog = new File(pipeLogDir, SyncConstant.getPipeLogName(startSerialNumber));
- createFile(newPipeLog);
-
- realTimeOutputStream = new DataOutputStream(new FileOutputStream(newPipeLog));
- realTimePipeLogStartNumber.offer(startSerialNumber);
- currentPipeLogSize = 0;
- }
-
- /** remove pipe log data */
- public void removePipeData(PipeData pipeData) throws IOException {
- long serialNumber = pipeData.getSerialNumber();
- serializeRemoveSerialNumber(serialNumber);
-
- // delete tsfile
- if (PipeData.Type.TSFILE.equals(pipeData.getType())) {
- List<File> tsFiles = ((TsFilePipeData) pipeData).getTsFiles();
- for (File file : tsFiles) {
- Files.deleteIfExists(file.toPath());
- }
- }
-
- // delete pipe log
- if (serialNumber >= 0) {
- if (historyOutputStream != null) {
- removeHistoryPipeLog();
- }
- if (realTimePipeLogStartNumber == null) {
- recoverRealTimePipeLogStartNumber();
- }
- if (realTimePipeLogStartNumber.size() >= 2) {
- long pipeLogStartNumber;
- while (true) {
- pipeLogStartNumber = realTimePipeLogStartNumber.poll();
- if (!realTimePipeLogStartNumber.isEmpty()
- && realTimePipeLogStartNumber.peek() < serialNumber) {
- removeRealTimePipeLog(pipeLogStartNumber);
- } else {
- break;
- }
- }
- realTimePipeLogStartNumber.addFirst(pipeLogStartNumber);
- }
- }
- }
-
- private void removeHistoryPipeLog() throws IOException {
- historyOutputStream.close();
- historyOutputStream = null;
- File historyPipeLog = new File(pipeLogDir, SyncConstant.HISTORY_PIPE_LOG_NAME);
- try {
- Files.delete(historyPipeLog.toPath());
- } catch (NoSuchFileException e) {
- logger.warn(
- String.format("delete history pipe log in %s error, %s", historyPipeLog.getPath(), e));
- }
- }
-
- private void removeRealTimePipeLog(long serialNumber) throws IOException {
- File realTimePipeLog = new File(pipeLogDir, SyncConstant.getPipeLogName(serialNumber));
- try {
- Files.delete(realTimePipeLog.toPath());
- } catch (NoSuchFileException e) {
- logger.warn(
- String.format("delete real time pipe log in %s error, %s", realTimePipeLog.getPath(), e));
- }
- }
-
- private void serializeRemoveSerialNumber(long serialNumber) throws IOException {
- if (removeSerialNumberWriter == null) {
- removeSerialNumberWriter =
- new BufferedWriter(new FileWriter(new File(pipeLogDir, SyncConstant.COMMIT_LOG_NAME)));
- currentRemoveLogSize = 0;
- }
- removeSerialNumberWriter.write(String.valueOf(serialNumber));
- removeSerialNumberWriter.newLine();
- removeSerialNumberWriter.flush();
- currentRemoveLogSize += Long.BYTES;
- if (currentRemoveLogSize >= SyncConstant.DEFAULT_PIPE_LOG_SIZE_IN_BYTE) {
- removeSerialNumberWriter.close();
- removeSerialNumberWriter = null;
- }
- }
-
public void finishCollect() {
try {
- if (createFile(new File(pipeDir, SyncConstant.FINISH_COLLECT_LOCK_NAME))) {
+ if (SyncPathUtil.createFile(new File(pipeDir, SyncConstant.FINISH_COLLECT_LOCK_NAME))) {
logger.info(String.format("Create finish collecting Lock file in %s.", pipeDir));
}
} catch (IOException e) {
@@ -307,27 +137,11 @@ public class TsFilePipeLogger {
}
}
- private boolean createFile(File file) throws IOException {
- if (!file.getParentFile().exists()) {
- file.getParentFile().mkdirs();
- }
- return file.createNewFile();
+ public boolean isCollectFinished() {
+ return new File(pipeDir, SyncConstant.FINISH_COLLECT_LOCK_NAME).exists();
}
public void clear() throws IOException {
- if (historyOutputStream != null) {
- removeHistoryPipeLog();
- }
- if (realTimeOutputStream != null) {
- realTimeOutputStream.close();
- realTimeOutputStream = null;
- }
- if (removeSerialNumberWriter != null) {
- removeSerialNumberWriter.close();
- removeSerialNumberWriter = null;
- }
-
- realTimePipeLogStartNumber = null;
File pipeDir = new File(this.pipeDir);
if (pipeDir.exists()) {
FileUtils.deleteDirectory(pipeDir);
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/sender/service/SenderService.java b/server/src/main/java/org/apache/iotdb/db/newsync/sender/service/SenderService.java
index fec7995..49eb25d 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/sender/service/SenderService.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/sender/service/SenderService.java
@@ -21,20 +21,28 @@ package org.apache.iotdb.db.newsync.sender.service;
import org.apache.iotdb.db.exception.PipeException;
import org.apache.iotdb.db.exception.PipeSinkException;
+import org.apache.iotdb.db.exception.ShutdownException;
import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.exception.SyncConnectionException;
import org.apache.iotdb.db.newsync.conf.SyncConstant;
import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
+import org.apache.iotdb.db.newsync.sender.pipe.IoTDBPipeSink;
import org.apache.iotdb.db.newsync.sender.pipe.Pipe;
import org.apache.iotdb.db.newsync.sender.pipe.PipeSink;
+import org.apache.iotdb.db.newsync.sender.pipe.TransportHandler;
import org.apache.iotdb.db.newsync.sender.pipe.TsFilePipe;
import org.apache.iotdb.db.newsync.sender.recovery.SenderLogAnalyzer;
import org.apache.iotdb.db.newsync.sender.recovery.SenderLogger;
+import org.apache.iotdb.db.newsync.transport.client.ITransportClient;
+import org.apache.iotdb.db.newsync.transport.client.TransportClient;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.physical.sys.CreatePipePlan;
import org.apache.iotdb.db.qp.physical.sys.CreatePipeSinkPlan;
import org.apache.iotdb.db.qp.utils.DatetimeUtils;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.db.utils.TestOnly;
+import org.apache.iotdb.service.transport.thrift.SyncResponse;
import org.apache.iotdb.tsfile.utils.Pair;
import org.slf4j.Logger;
@@ -49,20 +57,15 @@ import java.util.Map;
public class SenderService implements IService {
private static final Logger logger = LoggerFactory.getLogger(SenderService.class);
- private final SenderLogger senderLogger;
+ private SenderLogger senderLogger;
private Map<String, PipeSink> pipeSinks;
private List<Pipe> pipes;
private Pipe runningPipe;
+ private String runningMsg;
- private static volatile SenderService senderService;
-
- private SenderService() {
- pipeSinks = new HashMap<>();
- pipes = new ArrayList<>();
- senderLogger = new SenderLogger();
- }
+ private SenderService() {}
private static class SenderServiceHolder {
private static final SenderService INSTANCE = new SenderService();
@@ -138,7 +141,7 @@ public class SenderService implements IService {
}
/** pipe * */
- public void addPipe(CreatePipePlan plan) throws PipeException {
+ public synchronized void addPipe(CreatePipePlan plan) throws PipeException {
// common check
if (runningPipe != null && runningPipe.getStatus() != Pipe.PipeStatus.DROP) {
throw new PipeException(
@@ -175,19 +178,30 @@ public class SenderService implements IService {
}
}
- // get TsFilePipe
- // PipeSink.Type pipeSinkType = pipeSink.getType();
- // if (!pipeSinkType.equals(PipeSink.Type.IoTDB)) {
- // throw new PipeException(
- // String.format(
- // "Wrong pipeSink type %s for create TsFilePipe.", pipeSinkType)); // internal
- // error
- // }
- return new TsFilePipe(
- pipeCreateTime, plan.getPipeName(), pipeSink, plan.getDataStartTimestamp(), syncDelOp);
+ TsFilePipe pipe =
+ new TsFilePipe(
+ pipeCreateTime, plan.getPipeName(), pipeSink, plan.getDataStartTimestamp(), syncDelOp);
+ try {
+ if (!(pipeSink instanceof IoTDBPipeSink)) {
+ throw new PipeException(
+ String.format(
+ "Wrong pipeSink type %s for create pipe %s", pipeSink.getType(), pipe.getName()));
+ }
+ ITransportClient transportClient =
+ new TransportClient(
+ pipe, ((IoTDBPipeSink) pipeSink).getIp(), ((IoTDBPipeSink) pipeSink).getPort());
+ pipe.setTransportHandler(
+ new TransportHandler(transportClient, pipe.getName(), pipe.getCreateTime()));
+ } catch (IOException | SyncConnectionException e) {
+ e.printStackTrace();
+ throw new PipeException(
+ String.format(
+ "Create transport for pipe %s error, because %s.", pipe.getName(), e.getMessage()));
+ }
+ return pipe;
}
- public void stopPipe(String pipeName) throws PipeException {
+ public synchronized void stopPipe(String pipeName) throws PipeException {
checkRunningPipeExistAndName(pipeName);
if (runningPipe.getStatus() == Pipe.PipeStatus.RUNNING) {
runningPipe.stop();
@@ -195,7 +209,7 @@ public class SenderService implements IService {
senderLogger.operatePipe(pipeName, Operator.OperatorType.STOP_PIPE);
}
- public void startPipe(String pipeName) throws PipeException {
+ public synchronized void startPipe(String pipeName) throws PipeException {
checkRunningPipeExistAndName(pipeName);
if (runningPipe.getStatus() == Pipe.PipeStatus.STOP) {
runningPipe.start();
@@ -203,7 +217,7 @@ public class SenderService implements IService {
senderLogger.operatePipe(pipeName, Operator.OperatorType.START_PIPE);
}
- public void dropPipe(String pipeName) throws PipeException {
+ public synchronized void dropPipe(String pipeName) throws PipeException {
checkRunningPipeExistAndName(pipeName);
runningPipe.drop();
senderLogger.operatePipe(pipeName, Operator.OperatorType.DROP_PIPE);
@@ -213,6 +227,10 @@ public class SenderService implements IService {
return new ArrayList<>(pipes);
}
+ public synchronized String getPipeMsg(Pipe pipe) {
+ return pipe == runningPipe ? runningMsg : "";
+ }
+
private void checkRunningPipeExistAndName(String pipeName) throws PipeException {
if (runningPipe == null || runningPipe.getStatus() == Pipe.PipeStatus.DROP) {
throw new PipeException("There is no existing pipe.");
@@ -225,9 +243,48 @@ public class SenderService implements IService {
}
}
+ /** transport */
+ public synchronized void recMsg(SyncResponse response) {
+ if (runningPipe == null || runningPipe.getStatus() == Pipe.PipeStatus.DROP) {
+ logger.warn(String.format("No running pipe for receiving msg %s.", response));
+ return;
+ }
+ logger.info(String.format("%s from receiver: %s", response.type.name(), response.msg));
+ switch (response.type) {
+ case INFO:
+ break;
+ case ERROR:
+ try {
+ runningPipe.stop();
+ } catch (PipeException e) {
+ logger.error(
+ String.format(
+ "Stop pipe %s when meeting error in sender service, because %s.",
+ runningPipe.getName(), e));
+ }
+ case WARN:
+ if (runningMsg.length() > 0) {
+ runningMsg += System.lineSeparator();
+ }
+ runningMsg += (response.type.name() + " " + response.msg);
+ senderLogger.recordMsg(
+ runningPipe.getName(),
+ runningPipe.getStatus() == Pipe.PipeStatus.RUNNING
+ ? Operator.OperatorType.START_PIPE
+ : Operator.OperatorType.STOP_PIPE,
+ response.msg);
+ break;
+ }
+ }
+
/** IService * */
@Override
public void start() throws StartupException {
+ this.pipeSinks = new HashMap<>();
+ this.pipes = new ArrayList<>();
+ this.senderLogger = new SenderLogger();
+ this.runningMsg = "";
+
File senderLog = new File(SyncPathUtil.getSysDir(), SyncConstant.SENDER_LOG_NAME);
if (senderLog.exists()) {
try {
@@ -240,13 +297,35 @@ public class SenderService implements IService {
@Override
public void stop() {
- if (runningPipe != null) {
+ if (runningPipe != null && !Pipe.PipeStatus.DROP.equals(runningPipe.getStatus())) {
try {
runningPipe.stop();
} catch (PipeException e) {
- logger.warn(String.format("Running pipe %s has been dropped.", runningPipe.getName()));
+ logger.warn(
+ String.format(
+ "Stop pipe %s error when stop Sender Service, because %s.",
+ runningPipe.getName(), e));
}
}
+ }
+
+ @Override
+ public void shutdown(long milliseconds) throws ShutdownException {
+ if (runningPipe != null && !Pipe.PipeStatus.DROP.equals(runningPipe.getStatus())) {
+ try {
+ runningPipe.stop();
+ } catch (PipeException e) {
+ logger.warn(
+ String.format(
+ "Stop pipe %s error when shutdown Sender Service, because %s.",
+ runningPipe.getName(), e));
+ throw new ShutdownException(e);
+ }
+ }
+
+ pipeSinks = null;
+ pipes = null;
+ runningMsg = null;
senderLogger.close();
}
@@ -261,5 +340,12 @@ public class SenderService implements IService {
this.pipeSinks = analyzer.getRecoveryAllPipeSinks();
this.pipes = analyzer.getRecoveryAllPipes();
this.runningPipe = analyzer.getRecoveryRunningPipe();
+ this.runningMsg = analyzer.getRecoveryRunningMsg();
+ }
+
+ /** test */
+ @TestOnly
+ public Pipe getRunningPipe() {
+ return runningPipe;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java
index 1376dcc..c3c10e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java
@@ -19,4 +19,10 @@
*/
package org.apache.iotdb.db.newsync.transport.client;
-public interface ITransportClient {}
+import org.apache.iotdb.db.exception.SyncConnectionException;
+import org.apache.iotdb.service.transport.thrift.SyncRequest;
+import org.apache.iotdb.service.transport.thrift.SyncResponse;
+
+public interface ITransportClient extends Runnable {
+ SyncResponse heartbeat(SyncRequest syncRequest) throws SyncConnectionException;
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java
index 0179656..0c99bec 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java
@@ -23,16 +23,19 @@ import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.SyncConnectionException;
+import org.apache.iotdb.db.newsync.conf.SyncConstant;
+import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
import org.apache.iotdb.db.newsync.pipedata.PipeData;
import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
import org.apache.iotdb.db.newsync.sender.pipe.Pipe;
+import org.apache.iotdb.db.newsync.sender.service.SenderService;
import org.apache.iotdb.db.newsync.transport.conf.TransportConstant;
-import org.apache.iotdb.db.sync.conf.SyncSenderConfig;
-import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
+import org.apache.iotdb.db.newsync.transport.conf.TransportSenderConfig;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.rpc.RpcTransportFactory;
import org.apache.iotdb.service.transport.thrift.IdentityInfo;
import org.apache.iotdb.service.transport.thrift.MetaInfo;
+import org.apache.iotdb.service.transport.thrift.ResponseType;
import org.apache.iotdb.service.transport.thrift.SyncRequest;
import org.apache.iotdb.service.transport.thrift.SyncResponse;
import org.apache.iotdb.service.transport.thrift.TransportService;
@@ -44,16 +47,18 @@ import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
+import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
-import java.net.Socket;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
@@ -63,17 +68,19 @@ import static org.apache.iotdb.db.newsync.transport.conf.TransportConstant.REBAS
import static org.apache.iotdb.db.newsync.transport.conf.TransportConstant.RETRY_CODE;
import static org.apache.iotdb.db.newsync.transport.conf.TransportConstant.SUCCESS_CODE;
-public class TransportClient implements ITransportClient, Runnable {
+public class TransportClient implements ITransportClient {
private static final Logger logger = LoggerFactory.getLogger(TransportClient.class);
// TODO: Need to change to transport config
- private static SyncSenderConfig config = SyncSenderDescriptor.getInstance().getConfig();
+ private static TransportSenderConfig config = new TransportSenderConfig();
private static final IoTDBConfig ioTDBConfig = IoTDBDescriptor.getInstance().getConfig();
private static final int TIMEOUT_MS = 2000_000;
+ private static final int TRANSFER_BUFFER_SIZE_IN_BYTES = 1 * 1024 * 1024;
+
private TTransport transport = null;
private TransportService.Client serviceClient = null;
@@ -103,40 +110,43 @@ public class TransportClient implements ITransportClient, Runnable {
}
public TransportClient(Pipe pipe, String ipAddress, int port) throws IOException {
- Thread.currentThread().setName(ThreadName.SYNC_CLIENT.getName());
RpcTransportFactory.setThriftMaxFrameSize(ioTDBConfig.getThriftMaxFrameSize());
this.pipe = pipe;
this.ipAddress = ipAddress;
this.port = port;
-
- handshake();
}
- private boolean handshake() {
+ public boolean handshake() {
int handshakeCounter = 0;
- while (!handshakeWithVersion()) {
- handshakeCounter++;
- if (handshakeCounter > config.getMaxNumOfSyncFileRetry()) {
- logger.error(
+ try {
+ while (!handshakeWithVersion()) {
+ handshakeCounter++;
+ if (handshakeCounter > config.getMaxNumOfSyncFileRetry()) {
+ logger.error(
+ String.format(
+ "Handshake failed %s times! Check network.", config.getMaxNumOfSyncFileRetry()));
+ return false;
+ }
+ logger.info(
String.format(
- "Handshake failed %s times! Check network.", config.getMaxNumOfSyncFileRetry()));
- return false;
+ "Handshake error, retry %d/%d.",
+ handshakeCounter, config.getMaxNumOfSyncFileRetry()));
}
+ } catch (SyncConnectionException e) {
+ logger.error(String.format("Handshake failed and can not retry, because %s.", e));
+ return false;
}
return true;
}
- private boolean handshakeWithVersion() {
-
+ private boolean handshakeWithVersion() throws SyncConnectionException {
if (transport != null && transport.isOpen()) {
transport.close();
}
- try (Socket socket = new Socket(this.ipAddress, this.port)) {
- transport =
- RpcTransportFactory.INSTANCE.getTransport(
- config.getServerIp(), config.getServerPort(), TIMEOUT_MS);
+ try {
+ transport = RpcTransportFactory.INSTANCE.getTransport(ipAddress, port, TIMEOUT_MS);
TProtocol protocol;
if (ioTDBConfig.isRpcThriftCompressionEnable()) {
protocol = new TCompactProtocol(transport);
@@ -152,7 +162,7 @@ public class TransportClient implements ITransportClient, Runnable {
identityInfo =
new IdentityInfo(
- socket.getLocalAddress().getHostAddress(),
+ InetAddress.getLocalHost().getHostAddress(),
pipe.getName(),
pipe.getCreateTime(),
ioTDBConfig.getIoTDBMajorVersion());
@@ -161,19 +171,30 @@ public class TransportClient implements ITransportClient, Runnable {
throw new SyncConnectionException(
"The receiver rejected the synchronization task because " + status.msg);
}
- } catch (TTransportException e) {
- logger.error("Cannot connect to the receiver. ", e);
- // TODO: Do actions with exception.
- return false;
- } catch (SyncConnectionException | TException | IOException e) {
- logger.error("Cannot confirm identity with the receiver. ", e);
- // TODO: Do actions with exception.
+ } catch (TException e) {
+ logger.warn("Cannot connect to the receiver. ", e);
return false;
+ } catch (UnknownHostException e) {
+ logger.warn("Cannot confirm identity with the receiver. ", e);
+ throw new SyncConnectionException(String.format("Get local host error, because %s.", e));
}
return true;
}
- public boolean senderTransport(PipeData pipeData) {
+ public boolean senderTransport(PipeData pipeData) throws SyncConnectionException {
+ if (pipeData instanceof TsFilePipeData) {
+ try {
+ for (File file : ((TsFilePipeData) pipeData).getTsFiles(true)) {
+ transportSingleFile(file);
+ }
+ } catch (IOException e) {
+ logger.error(String.format("Get tsfiles error, because %s.", e));
+ return false;
+ } catch (NoSuchAlgorithmException e) {
+ logger.error(String.format("Wrong message digest, because %s.", e));
+ return false;
+ }
+ }
int retryCount = 0;
@@ -182,22 +203,26 @@ public class TransportClient implements ITransportClient, Runnable {
if (retryCount > config.getMaxNumOfSyncFileRetry()) {
logger.error(
String.format("After %s tries, stop the transport of current pipeData!", retryCount));
- return false;
+ throw new SyncConnectionException(
+ String.format("Can not connect to receiver when transferring pipedata %s.", pipeData));
}
try {
- if (pipeData instanceof TsFilePipeData) {
- for (File file : ((TsFilePipeData) pipeData).getTsFiles()) {
- transportSingleFile(file);
- }
- }
transportPipeData(pipeData);
logger.info("Finish current pipeData transport!");
break;
} catch (SyncConnectionException e) {
// handshake and retry
- handshake();
- } catch (IOException | NoSuchAlgorithmException e) {
+ if (!handshake()) {
+ logger.error(
+ String.format(
+ "Reconnect to receiver %s:%d error when transfer pipe data %s.",
+ ipAddress, port, pipeData));
+ throw new SyncConnectionException(
+ String.format(
+ "Reconnect to receiver error when transferring pipedata %s.", pipeData));
+ }
+ } catch (NoSuchAlgorithmException e) {
logger.error("Transport failed. ", e);
return false;
}
@@ -208,55 +233,42 @@ public class TransportClient implements ITransportClient, Runnable {
/** Transfer data of a tsfile to the receiver. */
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
private void transportSingleFile(File file)
- throws SyncConnectionException, IOException, NoSuchAlgorithmException {
+ throws SyncConnectionException, NoSuchAlgorithmException {
MessageDigest messageDigest = MessageDigest.getInstance("SHA-256");
+ int retryCount = 0;
while (true) {
- transportSingleFilePieceByPiece(file, messageDigest);
-
- if (isCheckFileDegistAgain) {
- // Check file digest as entirety.
- messageDigest.reset();
- try (InputStream inputStream = new FileInputStream(file)) {
- byte[] block = new byte[TransportConstant.DATA_CHUNK_SIZE];
- int length;
- while ((length = inputStream.read(block)) > 0) {
- messageDigest.update(block, 0, length);
- }
- }
- MetaInfo metaInfo = new MetaInfo(Type.FILE, file.getName(), 0);
-
- TransportStatus status = null;
+ retryCount++;
+ if (retryCount > config.getMaxNumOfSyncFileRetry()) {
+ throw new SyncConnectionException(
+ String.format("Connect to receiver error when transferring file %s.", file.getName()));
+ }
- int retryCount = 0;
+ try {
+ transportSingleFilePieceByPiece(file, messageDigest);
- while (true) {
- retryCount++;
- if (retryCount > config.getMaxNumOfSyncFileRetry()) {
- throw new SyncConnectionException(
- String.format(
- "Can not sync file %s after %s tries.",
- file.getAbsoluteFile(), config.getMaxNumOfSyncFileRetry()));
- }
+ if (isCheckFileDegistAgain) {
+ // Check file digest as entirety.
try {
- status =
- serviceClient.checkFileDigest(
- identityInfo, metaInfo, ByteBuffer.wrap(messageDigest.digest()));
- } catch (TException e) {
- // retry
- logger.error("TException happens! ", e);
- continue;
+ if (!checkFileDigest(file, messageDigest)) {
+ continue;
+ }
+ } catch (IOException e) {
+ logger.warn(
+ String.format(
+ "Read from disk to make digest error, skip check file %s, because %s.",
+ file.getName(), e));
}
- break;
}
-
- if (status.code != SUCCESS_CODE) {
- logger.error("Digest check of tsfile {} failed, retry", file.getAbsoluteFile());
- continue;
+ break;
+ } catch (SyncConnectionException e) {
+ if (!handshake()) {
+ throw new SyncConnectionException(
+ String.format(
+ "Connect to receiver error when transferring file %s.", file.getName()));
}
}
- break;
}
logger.info("Receiver has received {} successfully.", file.getAbsoluteFile());
@@ -268,8 +280,10 @@ public class TransportClient implements ITransportClient, Runnable {
// Cut the file into pieces to send
long position = 0;
+ long limit = getFileSizeLimit(file);
+
// Try small piece to rebase the file position.
- byte[] buffer = new byte[1024 * 1024];
+ byte[] buffer = new byte[TRANSFER_BUFFER_SIZE_IN_BYTES];
outer:
while (true) {
@@ -281,11 +295,13 @@ public class TransportClient implements ITransportClient, Runnable {
int dataLength;
try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
- if (randomAccessFile.length() <= position) {
+ if (limit <= position) {
break;
}
randomAccessFile.seek(position);
- while ((dataLength = randomAccessFile.read(buffer)) != -1) {
+ while ((dataLength =
+ randomAccessFile.read(buffer, 0, Math.min(buffer.length, (int) (limit - position))))
+ != -1) {
messageDigest.reset();
messageDigest.update(buffer, 0, dataLength);
ByteBuffer buffToSend = ByteBuffer.wrap(buffer, 0, dataLength);
@@ -327,9 +343,12 @@ public class TransportClient implements ITransportClient, Runnable {
"Receiver failed to receive data from {} because {}, abort.",
file.getAbsoluteFile(),
status.msg);
- throw new RuntimeException("Error! Replace this exception!");
+ throw new SyncConnectionException(status.msg);
} else { // Success
position += dataLength;
+ if (position >= limit) {
+ break;
+ }
}
}
} catch (IOException e) {
@@ -342,6 +361,61 @@ public class TransportClient implements ITransportClient, Runnable {
}
}
+ private long getFileSizeLimit(File file) {
+ File offset = new File(file.getPath() + SyncConstant.MODS_OFFSET_FILE_SUFFIX);
+ if (offset.exists()) {
+ try (BufferedReader br = new BufferedReader(new FileReader(offset))) {
+ return Long.parseLong(br.readLine());
+ } catch (IOException e) {
+ logger.error(
+ String.format("Deserialize offset of file %s error, because %s.", file.getPath(), e));
+ }
+ }
+ return file.length();
+ }
+
+ private boolean checkFileDigest(File file, MessageDigest messageDigest)
+ throws SyncConnectionException, IOException {
+ messageDigest.reset();
+ try (InputStream inputStream = new FileInputStream(file)) {
+ byte[] block = new byte[TransportConstant.DATA_CHUNK_SIZE];
+ int length;
+ while ((length = inputStream.read(block)) > 0) {
+ messageDigest.update(block, 0, length);
+ }
+ }
+
+ MetaInfo metaInfo = new MetaInfo(Type.FILE, file.getName(), 0);
+ TransportStatus status;
+ int retryCount = 0;
+
+ while (true) {
+ retryCount++;
+ if (retryCount > config.getMaxNumOfSyncFileRetry()) {
+ throw new SyncConnectionException(
+ String.format(
+ "Can not sync file %s after %s tries.",
+ file.getAbsoluteFile(), config.getMaxNumOfSyncFileRetry()));
+ }
+ try {
+ status =
+ serviceClient.checkFileDigest(
+ identityInfo, metaInfo, ByteBuffer.wrap(messageDigest.digest()));
+ } catch (TException e) {
+ // retry
+ logger.error("TException happens! ", e);
+ continue;
+ }
+ break;
+ }
+
+ if (status.code != SUCCESS_CODE) {
+ logger.error("Digest check of tsfile {} failed, retry", file.getAbsoluteFile());
+ return false;
+ }
+ return true;
+ }
+
private void transportPipeData(PipeData pipeData)
throws SyncConnectionException, NoSuchAlgorithmException {
@@ -393,30 +467,78 @@ public class TransportClient implements ITransportClient, Runnable {
*/
@Override
public void run() {
+ try {
+ if (!handshake()) {
+ throw new SyncConnectionException(
+ String.format("Handshake with receiver %s:%d error.", ipAddress, port));
+ }
+ while (!Thread.currentThread().isInterrupted()) {
+ PipeData pipeData = pipe.take();
+ if (!senderTransport(pipeData)) {
+ logger.warn(String.format("Can not transfer pipedata %s, skip it.", pipeData));
+ // can do something.
+ SenderService.getInstance()
+ .recMsg(
+ new SyncResponse(
+ ResponseType.WARN,
+ SyncPathUtil.createMsg(
+ String.format("Transfer piepdata %s error, skip it.", pipeData))));
+ continue;
+ }
+ pipe.commit();
+ }
+ } catch (InterruptedException e) {
+ logger.info("Interrupted by pipe, exit transport.");
+ } catch (SyncConnectionException e) {
+ logger.error(
+ String.format("Connect to receiver %s:%d error, because %s.", ipAddress, port, e));
+ SenderService.getInstance()
+ .recMsg(
+ new SyncResponse(
+ ResponseType.ERROR,
+ SyncPathUtil.createMsg(
+ String.format(
+ "Can not connect to %s:%d, because %s, please check receiver and internet.",
+ ipAddress, port, e.getMessage()))));
+ } finally {
+ close();
+ }
+ }
- // while (true) {
- // List<PipeData> pipeDataList = this.pipe.pull(Long.MAX_VALUE);
- // for (PipeData pipeData: pipeDataList) {
- // boolean sendResult = senderTransport(pipeData);
- // if (!sendResult) {
- // // Cannot handle the error.
- // return;
- // }
- // this.pipe.commit(pipeData.getSerialNumber());
- // }
- //
- // try {
- // Thread.sleep(1000);
- // } catch (InterruptedException e) {
- // Thread.currentThread().interrupt();
- // throw new RuntimeException(e);
- // }
- // }
+ @Override
+ public SyncResponse heartbeat(SyncRequest syncRequest) throws SyncConnectionException {
+ int retryCount = 0;
+ while (true) {
+ retryCount++;
+ if (retryCount > config.getMaxNumOfSyncFileRetry()) {
+ throw new SyncConnectionException(
+ String.format(
+ "%s request connects to receiver %s:%d error.",
+ syncRequest.type.name(), ipAddress, port));
+ }
- }
+ try (TTransport heartbeatTransport =
+ RpcTransportFactory.INSTANCE.getTransport(ipAddress, port, TIMEOUT_MS)) {
+ TProtocol protocol;
+ if (ioTDBConfig.isRpcThriftCompressionEnable()) {
+ protocol = new TCompactProtocol(heartbeatTransport);
+ } else {
+ protocol = new TBinaryProtocol(heartbeatTransport);
+ }
+ TransportService.Client heartbeatClient = new TransportService.Client(protocol);
+ if (!heartbeatTransport.isOpen()) {
+ heartbeatTransport.open();
+ }
- public SyncResponse heartbeat(SyncRequest syncRequest) throws TException {
- return serviceClient.heartbeat(identityInfo, syncRequest);
+ SyncResponse response = heartbeatClient.heartbeat(identityInfo, syncRequest);
+ return response;
+ } catch (TException e) {
+ logger.info(
+ String.format(
+ "Heartbeat connect to receiver %s:%d error, retry %d/%d.",
+ ipAddress, port, retryCount, config.getMaxNumOfSyncFileRetry()));
+ }
+ }
}
@TestOnly
@@ -447,7 +569,9 @@ public class TransportClient implements ITransportClient, Runnable {
}
public void close() {
- transport.close();
+ if (transport != null) {
+ transport.close();
+ }
}
private static class InstanceHolder {
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportSenderConfig.java b/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportSenderConfig.java
new file mode 100644
index 0000000..9d5b57e
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportSenderConfig.java
@@ -0,0 +1,129 @@
+/*
+ * 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.iotdb.db.newsync.transport.conf;
+
+import org.apache.iotdb.db.sync.conf.SyncConstant;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
+public class TransportSenderConfig {
+
+ private String serverIp = "127.0.0.1";
+
+ private int serverPort = 5555;
+
+ private int syncPeriodInSecond = 600;
+
+ private String senderFolderPath;
+
+ private String lastFileInfoPath;
+
+ private String snapshotPath;
+
+ /** The maximum number of retry when syncing a file to receiver fails. */
+ private int maxNumOfSyncFileRetry = 5;
+
+ /** Storage groups which participate in sync process */
+ private List<String> storageGroupList = new ArrayList<>();
+
+ /** Update paths based on data directory */
+ public void update(String dataDirectory) {
+ senderFolderPath =
+ dataDirectory
+ + File.separatorChar
+ + SyncConstant.SYNC_SENDER
+ + File.separatorChar
+ + getSyncReceiverName();
+ lastFileInfoPath = senderFolderPath + File.separatorChar + SyncConstant.LAST_LOCAL_FILE_NAME;
+ snapshotPath = senderFolderPath + File.separatorChar + SyncConstant.DATA_SNAPSHOT_NAME;
+ if (!new File(snapshotPath).exists()) {
+ new File(snapshotPath).mkdirs();
+ }
+ }
+
+ public String getServerIp() {
+ return serverIp;
+ }
+
+ public void setServerIp(String serverIp) {
+ this.serverIp = serverIp;
+ }
+
+ public int getServerPort() {
+ return serverPort;
+ }
+
+ public void setServerPort(int serverPort) {
+ this.serverPort = serverPort;
+ }
+
+ public int getSyncPeriodInSecond() {
+ return syncPeriodInSecond;
+ }
+
+ public void setSyncPeriodInSecond(int syncPeriodInSecond) {
+ this.syncPeriodInSecond = syncPeriodInSecond;
+ }
+
+ public String getSenderFolderPath() {
+ return senderFolderPath;
+ }
+
+ public void setSenderFolderPath(String senderFolderPath) {
+ this.senderFolderPath = senderFolderPath;
+ }
+
+ public String getLastFileInfoPath() {
+ return lastFileInfoPath;
+ }
+
+ public void setLastFileInfoPath(String lastFileInfoPath) {
+ this.lastFileInfoPath = lastFileInfoPath;
+ }
+
+ public String getSnapshotPath() {
+ return snapshotPath;
+ }
+
+ public void setSnapshotPath(String snapshotPath) {
+ this.snapshotPath = snapshotPath;
+ }
+
+ public String getSyncReceiverName() {
+ return serverIp + SyncConstant.SYNC_DIR_NAME_SEPARATOR + serverPort;
+ }
+
+ public List<String> getStorageGroupList() {
+ return new ArrayList<>(storageGroupList);
+ }
+
+ public void setStorageGroupList(List<String> storageGroupList) {
+ this.storageGroupList = storageGroupList;
+ }
+
+ public int getMaxNumOfSyncFileRetry() {
+ return maxNumOfSyncFileRetry;
+ }
+
+ public void setMaxNumOfSyncFileRetry(int maxNumOfSyncFileRetry) {
+ this.maxNumOfSyncFileRetry = maxNumOfSyncFileRetry;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java
index c9d8c64..6d00a82 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.newsync.conf.SyncConstant;
import org.apache.iotdb.db.service.ServiceType;
import org.apache.iotdb.db.service.thrift.ThriftService;
import org.apache.iotdb.db.service.thrift.ThriftServiceThread;
@@ -83,7 +84,7 @@ public class TransportServerManager extends ThriftService
getID().getName(),
ThreadName.SYNC_CLIENT.getName(),
config.getRpcAddress(),
- config.getSyncServerPort(),
+ SyncConstant.DEFAULT_PIPE_SINK_PORT,
Integer.MAX_VALUE,
config.getThriftServerAwaitTimeForStopService(),
new TransportServerThriftHandler(serviceImpl),
@@ -100,7 +101,7 @@ public class TransportServerManager extends ThriftService
@Override
public int getBindPort() {
// TODO: Whether to change this config here
- return IoTDBDescriptor.getInstance().getConfig().getSyncServerPort();
+ return SyncConstant.DEFAULT_PIPE_SINK_PORT;
}
// @Override
diff --git a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
index cdfa9db..e4998a0 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
import org.apache.iotdb.db.newsync.pipedata.PipeData;
import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueueFactory;
+import org.apache.iotdb.db.newsync.receiver.ReceiverService;
import org.apache.iotdb.service.transport.thrift.IdentityInfo;
import org.apache.iotdb.service.transport.thrift.MetaInfo;
import org.apache.iotdb.service.transport.thrift.SyncRequest;
@@ -273,9 +274,7 @@ public class TransportServiceImpl implements TransportService.Iface {
@Override
public SyncResponse heartbeat(IdentityInfo identityInfo, SyncRequest syncRequest)
throws TException {
-
- // return ReceiverService.getInstance().recMsg(syncRequest);
- return null;
+ return ReceiverService.getInstance().recMsg(syncRequest);
}
private void writeRecordFile(File recordFile, long position) throws IOException {
@@ -306,7 +305,6 @@ public class TransportServiceImpl implements TransportService.Iface {
File dir = new File(fileDir);
File[] targetFiles =
dir.listFiles((dir1, name) -> name.startsWith(tsFileName) && name.endsWith(PATCH_SUFFIX));
- // TODO: same name ?
if (targetFiles != null) {
for (File targetFile : targetFiles) {
File newFile =
@@ -319,6 +317,13 @@ public class TransportServiceImpl implements TransportService.Iface {
}
}
tsFilePipeData.setParentDirPath(dir.getAbsolutePath());
+ File recordFile = new File(fileDir, tsFileName + RECORD_SUFFIX);
+ try {
+ Files.deleteIfExists(recordFile.toPath());
+ } catch (IOException e) {
+ logger.warn(
+ String.format("Delete record file %s error, because %s.", recordFile.getPath(), e));
+ }
}
private String getFileDataDirPath(IdentityInfo identityInfo) {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 2946cd9..345d17b 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -220,6 +220,7 @@ import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PIPESINK_ATTRIBUTES;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PIPESINK_NAME;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PIPESINK_TYPE;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PIPE_CREATE_TIME;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PIPE_MSG;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PIPE_NAME;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PIPE_STATUS;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PRIVILEGE;
@@ -1200,8 +1201,14 @@ public class PlanExecutor implements IPlanExecutor {
new PartialPath(COLUMN_PIPE_CREATE_TIME, false),
new PartialPath(COLUMN_PIPE_NAME, false),
new PartialPath(COLUMN_PIPE2PIPESINK_NAME, false),
- new PartialPath(COLUMN_PIPE_STATUS, false)),
- Arrays.asList(TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT));
+ new PartialPath(COLUMN_PIPE_STATUS, false),
+ new PartialPath(COLUMN_PIPE_MSG, false)),
+ Arrays.asList(
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT));
boolean showAll = "".equals(plan.getPipeName());
for (Pipe pipe : SenderService.getInstance().getAllPipes())
if (showAll || plan.getPipeName().equals(pipe.getName())) {
@@ -1211,6 +1218,8 @@ public class PlanExecutor implements IPlanExecutor {
record.addField(Binary.valueOf(pipe.getName()), TSDataType.TEXT);
record.addField(Binary.valueOf(pipe.getPipeSink().getName()), TSDataType.TEXT);
record.addField(Binary.valueOf(pipe.getStatus().name()), TSDataType.TEXT);
+ record.addField(
+ Binary.valueOf(SenderService.getInstance().getPipeMsg(pipe)), TSDataType.TEXT);
listDataSet.putRecord(record);
}
return listDataSet;
@@ -2342,7 +2351,7 @@ public class PlanExecutor implements IPlanExecutor {
private void createPipe(CreatePipePlan plan) throws QueryProcessException {
try {
SenderService.getInstance().addPipe(plan);
- SenderService.getInstance().startPipe(plan.getPipeName());
+ // SenderService.getInstance().startPipe(plan.getPipeName());
} catch (PipeException e) {
throw new QueryProcessException("Create pipe error.", e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/EnvironmentUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
index fd6eb0b..941c526 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
@@ -338,6 +338,8 @@ public class EnvironmentUtils {
createDir(sgDir);
// create wal
createDir(config.getWalDir());
+ // create new sync
+ createDir(config.getNewSyncDir());
// create query
createDir(config.getQueryDir());
createDir(TestConstant.OUTPUT_DATA_DIR);
diff --git a/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java b/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java
index e187cce..1ab833b 100644
--- a/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java
@@ -244,7 +244,7 @@ public class BufferedPipeDataQueueTest {
BufferedPipeDataQueue pipeDataQueue = new BufferedPipeDataQueue(pipeLogDir.getPath());
Assert.assertEquals(1, pipeDataQueue.getCommitSerialNumber());
Assert.assertEquals(10, pipeDataQueue.getLastMaxSerialNumber());
- PipeData offerPipeData = new TsFilePipeData("", 11);
+ PipeData offerPipeData = new TsFilePipeData("fake11", 11);
pipeDataList.add(offerPipeData);
pipeDataQueue.offer(offerPipeData);
diff --git a/server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java b/server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
index 8e4d4d0..298316e 100644
--- a/server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.conf.SyncConstant;
import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
import org.apache.iotdb.db.newsync.pipedata.DeletionPipeData;
import org.apache.iotdb.db.newsync.pipedata.PipeData;
@@ -48,6 +49,7 @@ import org.junit.Before;
import org.junit.Test;
import java.io.*;
+import java.net.InetAddress;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.List;
@@ -59,15 +61,18 @@ public class TransportServiceTest {
File tmpDir = new File("target/synctest");
String pipeName1 = "pipe1";
- String remoteIp1 = "127.0.0.1";
+ String remoteIp1;
long createdTime1 = System.currentTimeMillis();
- File fileDir = new File(SyncPathUtil.getReceiverFileDataDir(pipeName1, remoteIp1, createdTime1));
- PipeDataQueue pipeDataQueue =
- PipeDataQueueFactory.getBufferedPipeDataQueue(
- SyncPathUtil.getReceiverPipeLogDir(pipeName1, remoteIp1, createdTime1));
+ File fileDir;
+ PipeDataQueue pipeDataQueue;
@Before
public void setUp() throws Exception {
+ remoteIp1 = InetAddress.getLocalHost().getHostAddress();
+ fileDir = new File(SyncPathUtil.getReceiverFileDataDir(pipeName1, remoteIp1, createdTime1));
+ pipeDataQueue =
+ PipeDataQueueFactory.getBufferedPipeDataQueue(
+ SyncPathUtil.getReceiverPipeLogDir(pipeName1, remoteIp1, createdTime1));
EnvironmentUtils.envSetUp();
if (!tmpDir.exists()) {
tmpDir.mkdirs();
@@ -121,7 +126,9 @@ public class TransportServiceTest {
// 4. start client
Pipe pipe = new TsFilePipe(createdTime1, pipeName1, null, 0, false);
- TransportClient client = new TransportClient(pipe, "127.0.0.1", 5555);
+ TransportClient client =
+ new TransportClient(pipe, "127.0.0.1", SyncConstant.DEFAULT_PIPE_SINK_PORT);
+ client.handshake();
for (PipeData pipeData : pipeDataList) {
client.senderTransport(pipeData);
}
@@ -168,6 +175,7 @@ public class TransportServiceTest {
for (int i = 0; i < resPipeData.size(); i++) {
Assert.assertEquals(pipeDataList.get(i), resPipeData.get(i));
}
+ pipeDataQueue.clear();
}
private void compareFile(File firFile, File secFile) {
diff --git a/server/src/test/resources/iotdb-engine.properties b/server/src/test/resources/iotdb-engine.properties
index a60bc73..2802cbd 100644
--- a/server/src/test/resources/iotdb-engine.properties
+++ b/server/src/test/resources/iotdb-engine.properties
@@ -24,3 +24,4 @@ index_root_dir=target/index
udf_root_dir=target/ext/udf
trigger_root_dir=target/ext/trigger
tracing_dir=target/data/tracing
+newsync_dir=target/newsync
\ No newline at end of file