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

Posted to notifications@asterixdb.apache.org by "Michael Blow (Code Review)" <do...@asterixdb.incubator.apache.org> on 2017/10/27 04:00:39 UTC

Change in asterixdb[master]: [NO ISSUE][*DB] LogFlusher fixes

Michael Blow has uploaded a new change for review.

  https://asterix-gerrit.ics.uci.edu/2106

Change subject: [NO ISSUE][*DB] LogFlusher fixes
......................................................................

[NO ISSUE][*DB] LogFlusher fixes

Change-Id: I19e150f2560573738938967f389a397ad7150a4d
---
M asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java
M asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
2 files changed, 61 insertions(+), 60 deletions(-)


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

diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java
index 081cf02..fa5e1bd 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java
@@ -105,15 +105,15 @@
                 if (logRecord.getLogType() == LogType.JOB_COMMIT || logRecord.getLogType() == LogType.ABORT
                         || logRecord.getLogType() == LogType.WAIT) {
                     logRecord.isFlushed(false);
-                    syncCommitQ.offer(logRecord);
+                    syncCommitQ.add(logRecord);
                 }
                 if (logRecord.getLogType() == LogType.FLUSH) {
                     logRecord.isFlushed(false);
-                    flushQ.offer(logRecord);
+                    flushQ.add(logRecord);
                 }
             } else if (logRecord.getLogSource() == LogSource.REMOTE
                     && (logRecord.getLogType() == LogType.JOB_COMMIT || logRecord.getLogType() == LogType.ABORT)) {
-                remoteJobsQ.offer(logRecord);
+                remoteJobsQ.add(logRecord);
             }
             this.notify();
         }
@@ -169,12 +169,13 @@
 
     @Override
     public void flush() {
+        boolean interrupted = false;
         try {
             int endOffset;
             while (!full.get()) {
-                synchronized (this) {
-                    if (appendOffset - flushOffset == 0 && !full.get()) {
-                        try {
+                try {
+                    synchronized (this) {
+                        if (appendOffset - flushOffset == 0 && !full.get()) {
                             if (IS_DEBUG_MODE) {
                                 LOGGER.info("flush()| appendOffset: " + appendOffset + ", flushOffset: " + flushOffset
                                         + ", full: " + full.get());
@@ -183,14 +184,14 @@
                                 fileChannel.close();
                                 return;
                             }
-                            this.wait();
-                        } catch (InterruptedException e) {
-                            continue;
+                            wait();
                         }
+                        endOffset = appendOffset;
                     }
-                    endOffset = appendOffset;
-                }
                 internalFlush(flushOffset, endOffset);
+                } catch (InterruptedException e) {
+                    interrupted = true;
+                }
             }
             internalFlush(flushOffset, appendOffset);
             if (isLastPage) {
@@ -198,6 +199,10 @@
             }
         } catch (IOException e) {
             throw new IllegalStateException(e);
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
         }
     }
 
@@ -230,7 +235,7 @@
         if (endOffset > beginOffset) {
             logBufferTailReader.initializeScan(beginOffset, endOffset);
 
-            ITransactionContext txnCtx = null;
+            ITransactionContext txnCtx;
 
             LogRecord logRecord = logBufferTailReader.next();
             while (logRecord != null) {
@@ -327,8 +332,9 @@
     }
 
     @Override
-    public void stop() {
-        this.stop = true;
+    public synchronized void stop() {
+        stop = true;
+        notifyAll();
     }
 
     @Override
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
index e5e91e8..1f45103 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
@@ -33,6 +33,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
@@ -162,7 +163,7 @@
             }
         }
 
-        /**
+        /*
          * To eliminate the case where the modulo of the next appendLSN = 0 (the next
          * appendLSN = the first LSN of the next log file), we do not allow a log to be
          * written at the last offset of the current file.
@@ -616,13 +617,11 @@
      * The deadlock happens when PrimaryIndexOpeartionTracker.completeOperation results in generating a FLUSH log and there are no empty log buffers available to log it.
      */
     private class FlushLogsLogger extends Thread {
-        private ILogRecord logRecord;
-
         @Override
         public void run() {
             while (true) {
                 try {
-                    logRecord = flushLogsQ.take();
+                    ILogRecord logRecord = flushLogsQ.take();
                     appendToLogTail(logRecord);
                 } catch (ACIDException e) {
                     e.printStackTrace();
@@ -641,77 +640,73 @@
     private final LinkedBlockingQueue<ILogBuffer> emptyQ;
     private final LinkedBlockingQueue<ILogBuffer> flushQ;
     private final LinkedBlockingQueue<ILogBuffer> stashQ;
-    private ILogBuffer flushPage;
-    private final AtomicBoolean isStarted;
-    private final AtomicBoolean terminateFlag;
+    private volatile ILogBuffer flushPage;
+    private final Semaphore started;
 
-    public LogFlusher(LogManager logMgr, LinkedBlockingQueue<ILogBuffer> emptyQ, LinkedBlockingQueue<ILogBuffer> flushQ,
+    LogFlusher(LogManager logMgr, LinkedBlockingQueue<ILogBuffer> emptyQ, LinkedBlockingQueue<ILogBuffer> flushQ,
             LinkedBlockingQueue<ILogBuffer> stashQ) {
         this.logMgr = logMgr;
         this.emptyQ = emptyQ;
         this.flushQ = flushQ;
         this.stashQ = stashQ;
-        flushPage = null;
-        isStarted = new AtomicBoolean(false);
-        terminateFlag = new AtomicBoolean(false);
-
+        this.started = new Semaphore(0);
     }
 
     public void terminate() {
         //make sure the LogFlusher thread started before terminating it.
-        synchronized (isStarted) {
-            while (!isStarted.get()) {
-                try {
-                    isStarted.wait();
-                } catch (InterruptedException e) {
-                    //ignore
-                }
+        boolean interrupted = false;
+        while (true) {
+            try {
+                started.acquire();
+                break;
+            } catch (InterruptedException e) { // NOSONAR- will re-interrupt after termination is processed
+                interrupted = true;
             }
         }
 
-        terminateFlag.set(true);
-        if (flushPage != null) {
-            synchronized (flushPage) {
-                flushPage.stop();
-                flushPage.notify();
+        final ILogBuffer currentFlushPage = flushPage;
+        if (currentFlushPage != null) {
+            currentFlushPage.stop();
+        }
+        while (true) {
+            try {
+                flushQ.put(POISON_PILL);
+                break;
+            } catch (InterruptedException e) { // NOSONAR- will re-interrupt after termination is processed
+                interrupted = true;
             }
         }
-        //[Notice]
-        //The return value doesn't need to be checked
-        //since terminateFlag will trigger termination if the flushQ is full.
-        flushQ.offer(POISON_PILL);
+        if (interrupted) {
+            Thread.currentThread().interrupt();
+        }
     }
 
     @Override
-    public Boolean call() {
-        synchronized (isStarted) {
-            isStarted.set(true);
-            isStarted.notify();
-        }
+    public Boolean call() throws InterruptedException {
+        started.release();
+        boolean interrupted = false;
         try {
             while (true) {
                 flushPage = null;
                 try {
                     flushPage = flushQ.take();
-                    if (flushPage == POISON_PILL || terminateFlag.get()) {
+                    if (flushPage == POISON_PILL) {
                         return true;
                     }
-                } catch (InterruptedException e) {
-                    if (flushPage == null) {
-                        continue;
-                    }
+                    flushPage.flush();
+                    // TODO(mblow): recycle large pages
+                    emptyQ.add(flushPage.getLogPageSize() == logMgr.getLogPageSize() ? flushPage : stashQ.remove());
+                } catch (InterruptedException e) { // NOSONAR: will rethrow after we have finished flushing
+                    interrupted = true;
                 }
-                flushPage.flush();
-                emptyQ.offer(flushPage.getLogPageSize() == logMgr.getLogPageSize() ? flushPage : stashQ.remove());
             }
         } catch (Exception e) {
-            if (LOGGER.isLoggable(Level.INFO)) {
-                LOGGER.info("-------------------------------------------------------------------------");
-                LOGGER.info("LogFlusher is terminating abnormally. System is in unusalbe state.");
-                LOGGER.info("-------------------------------------------------------------------------");
-            }
-            e.printStackTrace();
+            LOGGER.log(Level.SEVERE, "LogFlusher is terminating abnormally. System is in unusable state.", e);
             throw e;
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
         }
     }
 }

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

Gerrit-MessageType: newchange
Gerrit-Change-Id: I19e150f2560573738938967f389a397ad7150a4d
Gerrit-PatchSet: 1
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Owner: Michael Blow <mb...@apache.org>