You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/04/18 07:42:57 UTC
[GitHub] [iotdb] SilverNarcissus commented on a diff in pull request #5574: [To rel/0.13][IOTDB-2893] Double live for 0.13
SilverNarcissus commented on code in PR #5574:
URL: https://github.com/apache/iotdb/pull/5574#discussion_r851924966
##########
example/operationsync/src/main/java/org/apache/iotdb/operationsync/OperationSyncDegradationRate.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.operationsync;
+
+/**
+ * This is an operation sync insertion degradation rate test java class, which shows the performance
Review Comment:
We should move this file to test dir
##########
example/operationsync/src/main/java/org/apache/iotdb/operationsync/OperationSyncDurability.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.operationsync;
+
+import org.apache.iotdb.session.pool.SessionPool;
+
+public class OperationSyncDurability extends OperationSyncUtil {
Review Comment:
We should move this file to test dir
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProtector.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.writelog.io.SingleFileLogReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public abstract class OperationSyncProtector implements Runnable {
+
+ protected static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProtector.class);
+ protected static final int logFileValidity =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogValidity();
+
+ // For transmit log files
+ protected final Lock logFileListLock;
+ protected List<String> registeredLogFiles;
+ protected List<String> processingLogFiles;
+
+ // For serialize PhysicalPlan
+ private static final int MAX_PHYSICALPLAN_SIZE = 16 * 1024 * 1024;
+ protected final ByteArrayOutputStream protectorByteStream;
+ protected final DataOutputStream protectorDeserializeStream;
+
+ // Working state
+ protected volatile boolean isProtectorAtWork;
+
+ protected OperationSyncProtector() {
+ logFileListLock = new ReentrantLock();
+ registeredLogFiles = new ArrayList<>();
+
+ protectorByteStream = new ByteArrayOutputStream(MAX_PHYSICALPLAN_SIZE);
+ protectorDeserializeStream = new DataOutputStream(protectorByteStream);
+
+ isProtectorAtWork = false;
+ }
+
+ protected void registerLogFile(String logFile) {
+ logFileListLock.lock();
+ registeredLogFiles.add(logFile);
+ logFileListLock.unlock();
+ }
+
+ protected void wrapLogFiles() {
+ processingLogFiles = new ArrayList<>(registeredLogFiles);
+ registeredLogFiles = new ArrayList<>();
+ }
+
+ @Override
+ public void run() {
+ while (true) {
+ while (true) {
+ // Wrap and transmit all OperationSyncLogs
+ logFileListLock.lock();
+ if (registeredLogFiles.size() > 0) {
+ isProtectorAtWork = true;
+ wrapLogFiles();
+ logFileListLock.unlock();
Review Comment:
unlock should always be in finally block
##########
thrift/src/main/thrift/rpc.thrift:
##########
@@ -416,6 +416,12 @@ struct TSDropSchemaTemplateReq {
2: required string templateName
}
+struct TSOperationSyncWriteReq {
Review Comment:
We should also update "rpc-changelist.md" file
##########
server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java:
##########
@@ -790,6 +790,62 @@ private void loadProps() {
"iotdb_server_encrypt_decrypt_provider_parameter",
conf.getEncryptDecryptProviderParameter()));
+ // set OperationSync config
+ conf.setEnableOperationSync(
Review Comment:
Is this conf appeared at iotdb-engine.properties?
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncPlanTypeUtils.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
+import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
+
+public class OperationSyncPlanTypeUtils {
+
+ public static OperationSyncPlanType getOperationSyncPlanType(PhysicalPlan plan) {
+ if (plan instanceof SetStorageGroupPlan
+ || plan instanceof DeleteStorageGroupPlan
+ || plan instanceof CreateTimeSeriesPlan
+ || plan instanceof CreateMultiTimeSeriesPlan
+ || plan instanceof CreateAlignedTimeSeriesPlan
+ || plan instanceof DeleteTimeSeriesPlan
+ || plan instanceof AlterTimeSeriesPlan) {
+ return OperationSyncPlanType.EPlan;
+ } else if (plan instanceof DeletePlan) {
+ return OperationSyncPlanType.IPlan;
+ } else if (plan instanceof InsertPlan) {
+ return OperationSyncPlanType.NPlan;
+ }
+ return null;
+ }
+
+ public enum OperationSyncPlanType {
+ EPlan,
Review Comment:
Please give the meaning of E, I and N. Are they some sort of abbreviation？
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncLogService.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.writelog.io.LogWriter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class OperationSyncLogService implements Runnable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncLogService.class);
+
+ private static final String logFileDir =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogDir();
+ private static final long logFileValidity =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogValidity() * 1000L;
+ private static final int maxLogFileNum =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogNum();
+ private static final long maxLogFileSize =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncMaxLogSize();
+
+ private static long currentLogFileSize = 0;
+
+ private final OperationSyncProtector protector;
+ private final Lock logWriterLock;
+ private final String logFileName;
+ private int logFileID;
+ private long logFileCreateTime;
+ private File logFile;
+ private LogWriter logWriter;
+
+ public OperationSyncLogService(String logFileName, OperationSyncProtector protector) {
+ this.logFileName = logFileName;
+ this.protector = protector;
+
+ this.logWriterLock = new ReentrantLock();
+ this.logFile = null;
+ this.logWriter = null;
+
+ File logDir = new File(logFileDir);
+ if (!logDir.exists()) {
+ if (!logDir.mkdirs()) {
+ LOGGER.error("Can't make OperationSyncLog file dir: {}", logDir.getAbsolutePath());
+ }
+ }
+ }
+
+ @Override
+ public void run() {
+ // Check if there exists remnant logs
+ List<Integer> logFileIDList = new ArrayList<>();
+ for (int ID = 0; ID < maxLogFileNum; ID++) {
+ File file =
+ SystemFileFactory.INSTANCE.getFile(logFileDir + File.separator + logFileName + ID);
+ if (file.exists()) {
+ logFileIDList.add(ID);
+ }
+ }
+
+ int firstID = 0;
+ if (logFileIDList.size() > 0) {
+ // Re-transmit the remnant logs
+ for (int i = 0; i < logFileIDList.size() - 1; i++) {
+ if (logFileIDList.get(i + 1) - logFileIDList.get(i) > 1) {
+ firstID = i + 1;
+ break;
+ }
+ }
+
+ for (int i = firstID; i < logFileIDList.size(); i++) {
+ protector.registerLogFile(logFileDir + File.separator + logFileName + logFileIDList.get(i));
+ }
+ for (int i = 0; i < firstID; i++) {
+ protector.registerLogFile(logFileDir + File.separator + logFileName + logFileIDList.get(i));
+ }
+
+ int nextID;
+ if (firstID == 0) {
+ nextID = logFileIDList.get(logFileIDList.size() - 1) + 1;
+ } else {
+ nextID = logFileIDList.get(firstID - 1) + 1;
+ }
+ logFileID = nextID % maxLogFileNum;
+ } else {
+ logFileID = 0;
+ }
+
+ while (true) {
+ // Check the validity of logFile
+ logWriterLock.lock();
+ if (logWriter != null && System.currentTimeMillis() - logFileCreateTime > logFileValidity) {
+ // Submit logFile when it's expired
+ submitLogFile();
+ }
+ logWriterLock.unlock();
+
+ try {
+ // Sleep 10s before next check
+ TimeUnit.SECONDS.sleep(10);
+ } catch (InterruptedException e) {
+ LOGGER.error("OperationSyncLogService been interrupted", e);
+ }
+ }
+ }
+
+ private void submitLogFile() {
+ try {
+ logWriter.force();
+ } catch (IOException e) {
+ LOGGER.error("Can't force logWrite", e);
+ }
+ incLogFileSize(logFile.length());
+
+ for (int retry = 0; retry < 5; retry++) {
+ try {
+ logWriter.close();
+ } catch (IOException e) {
+ LOGGER.warn("Can't close OperationSyncLog: {}, retrying...", logFile.getAbsolutePath());
+ try {
+ // Sleep 1s and retry
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException ignored) {
+ // Ignore and retry
+ }
+ continue;
+ }
+
+ LOGGER.info("OperationSyncLog: {} is expired and closed", logFile.getAbsolutePath());
+ break;
+ }
+
+ protector.registerLogFile(
+ logFileDir
+ + File.separator
+ + logFileName
+ + (logFileID - 1 + maxLogFileNum) % maxLogFileNum);
+
+ logWriter = null;
+ logFile = null;
+ }
+
+ private void createLogFile() {
+ logFile =
+ SystemFileFactory.INSTANCE.getFile(logFileDir + File.separator + logFileName + logFileID);
+ while (true) {
+ try {
+ if (logFile.createNewFile()) {
+ logFileCreateTime = System.currentTimeMillis();
+ logWriter = new LogWriter(logFile, false);
+ LOGGER.info("Create OperationSyncLog: {}", logFile.getAbsolutePath());
+ break;
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Can't create OperationSyncLog: {}, retrying...", logFile.getAbsolutePath());
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException ignored) {
+ // Ignore and retry
+ }
+ }
+ }
+ logFileID = (logFileID + 1) % maxLogFileNum;
+ }
+
+ public static synchronized void incLogFileSize(long size) {
+ currentLogFileSize = currentLogFileSize + size;
+ }
+
+ public void acquireLogWriter() {
+ logWriterLock.lock();
+ }
+
+ public void write(ByteBuffer buffer) throws IOException {
+ if (currentLogFileSize < maxLogFileSize) {
+ if (logWriter == null) {
+ // Create logFile when there are no valid
+ createLogFile();
+ }
+ logWriter.write(buffer);
+ }
+ }
+
+ public void releaseLogWriter() {
Review Comment:
unlock should always be in a finally block
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncWriteTask.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.pool.SessionPool;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/** OperationSyncWriteTask is used for transmit one E-Plan sending by a client */
+public class OperationSyncWriteTask implements Runnable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncWriteTask.class);
+
+ private final ByteBuffer physicalPlanBuffer;
+ private final SessionPool operationSyncSessionPool;
+ private final OperationSyncDDLProtector eProtector;
+ private final OperationSyncLogService eLogService;
+
+ public OperationSyncWriteTask(
+ ByteBuffer physicalPlanBuffer,
+ SessionPool operationSyncSessionPool,
+ OperationSyncDDLProtector eProtector,
+ OperationSyncLogService eLogService) {
+ this.physicalPlanBuffer = physicalPlanBuffer;
+ this.operationSyncSessionPool = operationSyncSessionPool;
+ this.eProtector = eProtector;
+ this.eLogService = eLogService;
+ }
+
+ @Override
+ public void run() {
+ if (eProtector.isAtWork()) {
+ serializeEPlan();
+ } else {
+ boolean transmitStatus = false;
+ try {
+ physicalPlanBuffer.position(0);
+ transmitStatus = operationSyncSessionPool.operationSyncTransmit(physicalPlanBuffer);
+ } catch (IoTDBConnectionException connectionException) {
+ // warn IoTDBConnectionException and do serialization
+ LOGGER.warn(
+ "OperationSyncWriteTask can't transmit because network failure", connectionException);
+ } catch (Exception e) {
+ // The PhysicalPlan has internal error, reject transmit
+ LOGGER.error("OperationSyncWriteTask can't transmit", e);
+ return;
+ }
+ if (!transmitStatus) {
+ serializeEPlan();
+ }
+ }
+ }
+
+ private void serializeEPlan() {
+ // serialize the E-Plan if necessary
+ try {
+ // must set buffer position to limit() before serialization
+ physicalPlanBuffer.position(physicalPlanBuffer.limit());
+ eLogService.acquireLogWriter();
+ eLogService.write(physicalPlanBuffer);
+ } catch (IOException e) {
+ LOGGER.error("can't serialize current PhysicalPlan", e);
+ }
+ eLogService.releaseLogWriter();
Review Comment:
unlock should be in finally block.
##########
example/operationsync/src/main/java/org/apache/iotdb/operationsync/OperationSyncEmpty.java:
##########
@@ -0,0 +1,26 @@
+/*
+ * 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.operationsync;
+
+public class OperationSyncEmpty extends OperationSyncUtil {
Review Comment:
We should move this file to test dir
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncLogService.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.writelog.io.LogWriter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class OperationSyncLogService implements Runnable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncLogService.class);
+
+ private static final String logFileDir =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogDir();
+ private static final long logFileValidity =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogValidity() * 1000L;
+ private static final int maxLogFileNum =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogNum();
+ private static final long maxLogFileSize =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncMaxLogSize();
+
+ private static long currentLogFileSize = 0;
+
+ private final OperationSyncProtector protector;
+ private final Lock logWriterLock;
+ private final String logFileName;
+ private int logFileID;
+ private long logFileCreateTime;
+ private File logFile;
+ private LogWriter logWriter;
+
+ public OperationSyncLogService(String logFileName, OperationSyncProtector protector) {
+ this.logFileName = logFileName;
+ this.protector = protector;
+
+ this.logWriterLock = new ReentrantLock();
+ this.logFile = null;
+ this.logWriter = null;
+
+ File logDir = new File(logFileDir);
+ if (!logDir.exists()) {
+ if (!logDir.mkdirs()) {
+ LOGGER.error("Can't make OperationSyncLog file dir: {}", logDir.getAbsolutePath());
+ }
+ }
+ }
+
+ @Override
+ public void run() {
+ // Check if there exists remnant logs
+ List<Integer> logFileIDList = new ArrayList<>();
+ for (int ID = 0; ID < maxLogFileNum; ID++) {
+ File file =
+ SystemFileFactory.INSTANCE.getFile(logFileDir + File.separator + logFileName + ID);
+ if (file.exists()) {
+ logFileIDList.add(ID);
+ }
+ }
+
+ int firstID = 0;
+ if (logFileIDList.size() > 0) {
+ // Re-transmit the remnant logs
+ for (int i = 0; i < logFileIDList.size() - 1; i++) {
+ if (logFileIDList.get(i + 1) - logFileIDList.get(i) > 1) {
+ firstID = i + 1;
+ break;
+ }
+ }
+
+ for (int i = firstID; i < logFileIDList.size(); i++) {
+ protector.registerLogFile(logFileDir + File.separator + logFileName + logFileIDList.get(i));
+ }
+ for (int i = 0; i < firstID; i++) {
+ protector.registerLogFile(logFileDir + File.separator + logFileName + logFileIDList.get(i));
+ }
+
+ int nextID;
+ if (firstID == 0) {
+ nextID = logFileIDList.get(logFileIDList.size() - 1) + 1;
+ } else {
+ nextID = logFileIDList.get(firstID - 1) + 1;
+ }
+ logFileID = nextID % maxLogFileNum;
+ } else {
+ logFileID = 0;
+ }
+
+ while (true) {
+ // Check the validity of logFile
+ logWriterLock.lock();
+ if (logWriter != null && System.currentTimeMillis() - logFileCreateTime > logFileValidity) {
+ // Submit logFile when it's expired
+ submitLogFile();
+ }
+ logWriterLock.unlock();
Review Comment:
unlock should be in finally block.
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncDDLProtector.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.pool.SessionPool;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
+
+public class OperationSyncDDLProtector extends OperationSyncProtector {
Review Comment:
Since there is only one DDLProtector, this class should be a singleton
##########
server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java:
##########
@@ -302,9 +316,63 @@ public TSFetchResultsResp call() throws Exception {
protected final ServiceProvider serviceProvider;
+ /* OperationSync module */
+ private static final boolean isEnableOperationSync =
+ IoTDBDescriptor.getInstance().getConfig().isEnableOperationSync();
+ private final SessionPool operationSyncsessionPool;
+ private final OperationSyncProducer operationSyncProducer;
+ private final OperationSyncDDLProtector operationSyncDDLProtector;
+ private final OperationSyncLogService operationSyncDDLLogService;
+
public TSServiceImpl() {
super();
serviceProvider = IoTDB.serviceProvider;
+
+ if (isEnableOperationSync) {
+ /* Open OperationSync */
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ // create SessionPool for OperationSync
+ operationSyncsessionPool =
+ new SessionPool(
+ config.getSecondaryAddress(),
+ config.getSecondaryPort(),
+ config.getSecondaryUser(),
+ config.getSecondaryPassword(),
+ 5);
+
+ // create operationSyncDDLProtector and operationSyncDDLLogService
+ operationSyncDDLProtector = new OperationSyncDDLProtector(operationSyncsessionPool);
+ new Thread(operationSyncDDLProtector).start();
+ operationSyncDDLLogService =
+ new OperationSyncLogService("OperationSyncDDLLog", operationSyncDDLProtector);
+ new Thread(operationSyncDDLLogService).start();
+
+ // create OperationSyncProducer
+ BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
+ blockingQueue = new ArrayBlockingQueue<>(config.getOperationSyncProducerCacheSize());
+ operationSyncProducer = new OperationSyncProducer(blockingQueue);
+
+ // create OperationSyncDMLProtector and OperationSyncDMLLogService
+ OperationSyncDMLProtector operationSyncDMLProtector =
+ new OperationSyncDMLProtector(operationSyncDDLProtector, operationSyncProducer);
+ new Thread(operationSyncDMLProtector).start();
+ OperationSyncLogService operationSyncDMLLogService =
+ new OperationSyncLogService("OperationSyncDMLLog", operationSyncDMLProtector);
+ new Thread(operationSyncDMLLogService).start();
+
+ // create OperationSyncConsumer
+ for (int i = 0; i < config.getOperationSyncConsumerConcurrencySize(); i++) {
+ OperationSyncConsumer consumer =
+ new OperationSyncConsumer(
+ blockingQueue, operationSyncsessionPool, operationSyncDMLLogService);
+ new Thread(consumer).start();
+ }
+ } else {
+ operationSyncsessionPool = null;
Review Comment:
The default value of them is null. We can remove this else block.
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProtector.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.writelog.io.SingleFileLogReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public abstract class OperationSyncProtector implements Runnable {
+
+ protected static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProtector.class);
+ protected static final int logFileValidity =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogValidity();
+
+ // For transmit log files
+ protected final Lock logFileListLock;
+ protected List<String> registeredLogFiles;
+ protected List<String> processingLogFiles;
+
+ // For serialize PhysicalPlan
+ private static final int MAX_PHYSICALPLAN_SIZE = 16 * 1024 * 1024;
+ protected final ByteArrayOutputStream protectorByteStream;
+ protected final DataOutputStream protectorDeserializeStream;
+
+ // Working state
+ protected volatile boolean isProtectorAtWork;
+
+ protected OperationSyncProtector() {
+ logFileListLock = new ReentrantLock();
+ registeredLogFiles = new ArrayList<>();
+
+ protectorByteStream = new ByteArrayOutputStream(MAX_PHYSICALPLAN_SIZE);
+ protectorDeserializeStream = new DataOutputStream(protectorByteStream);
+
+ isProtectorAtWork = false;
+ }
+
+ protected void registerLogFile(String logFile) {
+ logFileListLock.lock();
+ registeredLogFiles.add(logFile);
+ logFileListLock.unlock();
Review Comment:
unlock should be in finally block.
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncLogService.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.writelog.io.LogWriter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class OperationSyncLogService implements Runnable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncLogService.class);
+
+ private static final String logFileDir =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogDir();
+ private static final long logFileValidity =
+ IoTDBDescriptor.getInstance().getConfig().getOperationSyncLogValidity() * 1000L;
Review Comment:
Why we multiply 1000 here? This will cause the default value of logFileValidity become 30000 and we will sleep 30000 seconds later. Please double check here.
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncDMLProtector.java:
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.doublelive;
+
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
+
+public class OperationSyncDMLProtector extends OperationSyncProtector {
Review Comment:
Since there is only one DMLProtector, this class should be a singleton
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org