Merge pull request #2302 from adamretter/hotfix/journal-must-write-ahead

Journal must write-ahead

Author: wolfgangmm

src/org/exist/storage/BrokerPool.java

@@ -22,6 +22,7 @@
 import com.evolvedbinary.j8fu.fsm.AtomicFSM;
 import com.evolvedbinary.j8fu.fsm.FSM;
 import net.jcip.annotations.GuardedBy;
+import net.jcip.annotations.NotThreadSafe;
 import net.jcip.annotations.ThreadSafe;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -1409,16 +1410,25 @@ public long getLastMajorSync() {
     /**
      * Executes a waiting cache synchronization for the database instance.
      *
+     * NOTE: This method should not be called concurrently from multiple threads.
+     *
      * @param broker    A broker responsible for executing the job
      * @param syncEvent One of {@link org.exist.storage.sync.Sync}
      */
-    //TODO : rename as runSync ? executeSync ?
-    //TOUNDERSTAND (pb) : *not* synchronized, so... "executes" or, rather, "schedules" ? "executes" (WM)
-    //TOUNDERSTAND (pb) : why do we need a broker here ? Why not get and release one when we're done ?
-    // WM: the method will always be under control of the BrokerPool. It is guaranteed that no
-    // other brokers are active when it is called. That's why we don't need to synchronize here.
-    //TODO : make it protected ?
     public void sync(final DBBroker broker, final Sync syncEvent) {
+
+        /**
+         * Database Systems - The Complete Book (Second edition)
+         * § 17.4.1 The Undo/Redo Rules
+         *
+         * The constraints that an undo/redo logging system must follow are summarized by the following rule:
+         *     * UR1  Before modifying any database element X on disk because of changes
+         *            made by some transaction T, it is necessary that the update record
+         *            <T,X,v,w> appear on disk.
+         */
+        journalManager.get().flush(true, true);
+
+        // sync various DBX files
         broker.sync(syncEvent);
 
         //TODO : strange that it is set *after* the sunc method has been called.

src/org/exist/storage/journal/FileSyncRunnable.java

@@ -98,7 +98,7 @@
  */
 @ConfigurationClass("journal")
 //TODO: conf.xml refactoring <recovery> => <recovery><journal/></recovery>
-public final class Journal {
+public final class Journal implements Closeable {
     /**
      * Logger for this class
      */
@@ -187,13 +187,7 @@ public final class Journal {
      * the current output channel
      * Only valid after switchFiles() was called at least once!
      */
-    private SeekableByteChannel channel;
-
-    /**
-     * Syncing the journal is done by a background thread
-     */
-    private final FileSyncRunnable fileSyncRunnable;
-    private final Thread fileSyncThread;
+    private FileChannel channel;
 
     /**
      * latch used to synchronize writes to the channel
@@ -263,10 +257,6 @@ public Journal(final BrokerPool pool, final Path directory) throws EXistExceptio
         this.fsJournalDir = directory.resolve("fs.journal");
         this.currentBuffer = ByteBuffer.allocateDirect(BUFFER_SIZE);
 
-        this.fileSyncRunnable = new FileSyncRunnable(latch);
-        this.fileSyncThread = newInstanceThread(pool, "file-sync-thread", fileSyncRunnable);
-        fileSyncThread.start(); //this makes us to use class as a final only - no inheritance allowed
-
         this.syncOnCommit = pool.getConfiguration().getProperty(PROPERTY_RECOVERY_SYNC_ON_COMMIT, DEFAULT_SYNC_ON_COMMIT);
         if (LOG.isDebugEnabled()) {
             LOG.debug("SyncOnCommit = " + syncOnCommit);
@@ -409,11 +399,18 @@ public synchronized void flushToLog(final boolean fsync, final boolean forceSync
         if (inRecovery) {
             return;
         }
+
         flushBuffer();
-        if (forceSync || (fsync && syncOnCommit && currentLsn.compareTo(lastSyncLsn) > 0)) {
-            fileSyncRunnable.triggerSync();
-            lastSyncLsn = currentLsn;
+
+        try {
+            if (forceSync || (fsync && syncOnCommit && currentLsn.compareTo(lastSyncLsn) > 0)) {
+                sync();
+                lastSyncLsn = currentLsn;
+            }
+        } catch (final IOException e) {
+            LOG.error("Could not sync Journal to disk: " + e.getMessage(), e);
         }
+
         try {
             if (channel != null && channel.size() >= journalSizeLimit) {
                 pool.triggerCheckpoint();
@@ -423,6 +420,10 @@ public synchronized void flushToLog(final boolean fsync, final boolean forceSync
         }
     }
 
+    private void sync() throws IOException {
+        channel.force(true);
+    }
+
     /**
      * Flush the buffer to disk.
      */
@@ -541,11 +542,13 @@ public void switchFiles() throws LogException {
         }
 
         synchronized (latch) {
-            close();
             try {
-                channel = Files.newByteChannel(file, CREATE_NEW, WRITE);
+                // close current file
+                close();
+
+                // open new file
+                channel = (FileChannel) Files.newByteChannel(file, CREATE_NEW, WRITE);
                 writeJournalHeader(channel);
-                fileSyncRunnable.setChannel((FileChannel) channel);
                 initialised = true;
             } catch (final IOException e) {
                 throw new LogException("Failed to open new journal: " + file.toAbsolutePath().toString(), e);
@@ -571,13 +574,15 @@ private void writeJournalHeader(final SeekableByteChannel channel) throws IOExce
     /**
      * Close the journal.
      */
-    public void close() {
+    @Override
+    public void close() throws IOException {
         if (channel != null) {
             try {
-                channel.close();
+                sync();
             } catch (final IOException e) {
-                LOG.warn("Failed to close journal channel", e);
+                LOG.error(e.getMessage(), e);
             }
+            channel.close();
         }
     }
 
@@ -647,23 +652,23 @@ public void shutdown(final long txnId, final boolean checkpoint) {
 
         if (!BrokerPool.FORCE_CORRUPTION) {
             if (checkpoint) {
-                LOG.info("Transaction journal cleanly shutting down with checkpoint...");
+                LOG.info("Shutting down Journal with checkpoint...");
                 try {
                     writeToLog(new Checkpoint(txnId));
                 } catch (final JournalException e) {
-                    LOG.error("An error occurred while closing the journal file: " + e.getMessage(), e);
+                    LOG.error("An error occurred whilst writing a checkpoint to the Journal: " + e.getMessage(), e);
                 }
             }
             flushBuffer();
         }
-        fileLock.release();
-        fileSyncRunnable.shutdown();
-        fileSyncThread.interrupt();
+
         try {
-            fileSyncThread.join();
-        } catch (final InterruptedException e) {
-            //Nothing to do
+            channel.close();
+        } catch (final IOException e) {
+            LOG.error("Unable to close Journal file: " + e.getMessage(), e);
         }
+        channel = null;
+        fileLock.release();
         currentBuffer = null;
     }