You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@asterixdb.apache.org by "Murtadha Hubail (Code Review)" <de...@asterixdb.apache.org> on 2019/06/28 01:39:30 UTC

Change in asterixdb[stabilization-f69489]: [NO ISSUE][REPL] Suspend Dataset Checkpointing on Replica Sync

Murtadha Hubail has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/3469


Change subject: [NO ISSUE][REPL] Suspend Dataset Checkpointing on Replica Sync
......................................................................

[NO ISSUE][REPL] Suspend Dataset Checkpointing on Replica Sync

- user model changes: no
- storage format changes: no
- interface changes: yes

Details:
- Before synchronizing replicas, stop datasets checkpointing to
  prevent new files from being generated due to async IO operations
  triggered by checkpointing.
- Instead of sync'ing current files to replicas then scheduling a flush
  and sync'ing any newly generated files, just flush datasets before
  the initial sync then sync all the files in one go.

Change-Id: I058fd48bc0fb89a1e16448ce516c3410bb4d681d
---
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ICheckpointManager.java
M asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
M asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointManager.java
3 files changed, 37 insertions(+), 11 deletions(-)



  git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/69/3469/1

diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ICheckpointManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ICheckpointManager.java
index 36cea55..954e399 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ICheckpointManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ICheckpointManager.java
@@ -58,4 +58,14 @@
      * @param id
      */
     void completed(TxnId id);
+
+    /**
+     * Suspends checkpointing datasets
+     */
+    void suspend();
+
+    /**
+     * Resumes checkpointing datasets
+     */
+    void resume();
 }
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
index 0f0b5bd..123709b 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
@@ -22,6 +22,7 @@
 
 import org.apache.asterix.common.api.INcApplicationContext;
 import org.apache.asterix.common.replication.IReplicationStrategy;
+import org.apache.asterix.common.transactions.ICheckpointManager;
 import org.apache.asterix.replication.api.PartitionReplica;
 import org.apache.asterix.replication.messaging.CheckpointPartitionIndexesTask;
 import org.apache.asterix.replication.messaging.ReplicationProtocol;
@@ -45,21 +46,25 @@
     public void sync() throws IOException {
         final Object syncLock = appCtx.getReplicaManager().getReplicaSyncLock();
         synchronized (syncLock) {
-            syncFiles();
-            checkpointReplicaIndexes();
-            appCtx.getReplicationManager().register(replica);
+            final ICheckpointManager checkpointManager = appCtx.getTransactionSubsystem().getCheckpointManager();
+            try {
+                // suspend checkpointing datasets to prevent async IO operations while sync'ing replicas
+                checkpointManager.suspend();
+                syncFiles();
+                checkpointReplicaIndexes();
+                appCtx.getReplicationManager().register(replica);
+            } finally {
+                checkpointManager.resume();
+            }
         }
     }
 
     private void syncFiles() throws IOException {
         final ReplicaFilesSynchronizer fileSync = new ReplicaFilesSynchronizer(appCtx, replica);
-        waitForReplicatedDatasetsIO();
-        fileSync.sync();
         // flush replicated dataset to generate disk component for any remaining in-memory components
         final IReplicationStrategy replStrategy = appCtx.getReplicationManager().getReplicationStrategy();
         appCtx.getDatasetLifecycleManager().flushDataset(replStrategy);
         waitForReplicatedDatasetsIO();
-        // sync any newly generated files
         fileSync.sync();
     }
 
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointManager.java
index ce523db..b85742e 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointManager.java
@@ -18,6 +18,10 @@
  */
 package org.apache.asterix.transaction.management.service.recovery;
 
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
 import org.apache.asterix.common.transactions.CheckpointProperties;
 import org.apache.asterix.common.transactions.ICheckpointManager;
@@ -26,10 +30,6 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
 
 /**
  * An implementation of {@link ICheckpointManager} that defines the logic
@@ -40,6 +40,7 @@
     private static final Logger LOGGER = LogManager.getLogger();
     private static final long NO_SECURED_LSN = -1L;
     private final Map<TxnId, Long> securedLSNs;
+    private boolean suspended = false;
 
     public CheckpointManager(ITransactionSubsystem txnSubsystem, CheckpointProperties checkpointProperties) {
         super(txnSubsystem, checkpointProperties);
@@ -76,7 +77,7 @@
         }
         final long minFirstLSN = txnSubsystem.getRecoveryManager().getMinFirstLSN();
         boolean checkpointSucceeded = minFirstLSN >= checkpointTargetLSN;
-        if (!checkpointSucceeded) {
+        if (!checkpointSucceeded && !suspended) {
             // Flush datasets with indexes behind target checkpoint LSN
             IDatasetLifecycleManager datasetLifecycleManager =
                     txnSubsystem.getApplicationContext().getDatasetLifecycleManager();
@@ -100,6 +101,16 @@
         securedLSNs.remove(id);
     }
 
+    @Override
+    public synchronized void suspend() {
+        suspended = true;
+    }
+
+    @Override
+    public synchronized void resume() {
+        suspended = false;
+    }
+
     private synchronized long getMinSecuredLSN() {
         return securedLSNs.isEmpty() ? NO_SECURED_LSN : Collections.min(securedLSNs.values());
     }

-- 
To view, visit https://asterix-gerrit.ics.uci.edu/3469
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings

Gerrit-Project: asterixdb
Gerrit-Branch: stabilization-f69489
Gerrit-MessageType: newchange
Gerrit-Change-Id: I058fd48bc0fb89a1e16448ce516c3410bb4d681d
Gerrit-Change-Number: 3469
Gerrit-PatchSet: 1
Gerrit-Owner: Murtadha Hubail <mh...@apache.org>