Implements interrupt-safe transaction log

When an interrupt happened on a thread during writing the transaction
log, the underlying FileChannel was closed as a result. Any further
use (even from another threads) threw a
java.nio.channels.ClosedChannelException rendering the transaction
manager unusable since the transaction logs were never reopened. This
commit changes this behavior, log file operations reopen the file if
it was closed by an interrupt of another thread.

fixes scalar-labs#45

Commit by Tibor Billes, Miklós Karakó, Balázs Póka

Author: billestibor

btm/pom.xml

@@ -12,7 +12,7 @@
     <packaging>bundle</packaging>
 
     <dependencies>
-        <dependency>
+	    <dependency>
             <groupId>javax.transaction</groupId>
             <artifactId>jta</artifactId>
             <scope>provided</scope>
@@ -70,6 +70,24 @@
             <scope>provided</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>1.3.2</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>3.4</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.easytesting</groupId>
+            <artifactId>fest-assert-core</artifactId>
+            <version>2.0M10</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

btm/src/main/java/bitronix/tm/gui/TransactionLogHeaderPanel.java

@@ -16,6 +16,7 @@
 package bitronix.tm.gui;
 
 import bitronix.tm.journal.TransactionLogHeader;
+import bitronix.tm.journal.InterruptibleLockedRandomAccessFile;
 import bitronix.tm.utils.Decoder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -74,8 +75,8 @@ public void setPosition(long position) {
     }
 
     public void read(File logFile, boolean active) throws IOException {
-        RandomAccessFile raf = new RandomAccessFile(logFile, "r");
-        TransactionLogHeader header = new TransactionLogHeader(raf.getChannel(), 0L);
+        InterruptibleLockedRandomAccessFile raf = new InterruptibleLockedRandomAccessFile(logFile, "r");
+        TransactionLogHeader header = new TransactionLogHeader(raf, 0L);
         raf.close();
         if (log.isDebugEnabled()) { log.debug("read header: " + header); }
         setLogFile(logFile);

btm/src/main/java/bitronix/tm/journal/InterruptibleLockedRandomAccessFile.java

@@ -0,0 +1,104 @@
+package bitronix.tm.journal;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+
+public class InterruptibleLockedRandomAccessFile {
+
+    private final File file;
+    private final String mode;
+    private RandomAccessFile openedFile;
+    private FileChannel fileChannel;
+    private FileLock fileLock;
+    private long currentPosition = 0;
+    private boolean closed;
+
+    public InterruptibleLockedRandomAccessFile(final File file, final String mode)
+            throws IOException {
+        this.file = file;
+        this.mode = mode;
+        open();
+    }
+
+    private synchronized void open() throws IOException {
+        openedFile = new RandomAccessFile(file, mode);
+        fileChannel = openedFile.getChannel();
+
+        final boolean shared = false;
+        this.fileLock = fileChannel
+                .tryLock(0, TransactionLogHeader.TIMESTAMP_HEADER, shared);
+        if (this.fileLock == null) {
+            throw new IOException("File " + file.getAbsolutePath()
+                    + " is locked. Is another instance already running?");
+        }
+    }
+
+    public synchronized final void close() throws IOException {
+        try {
+            if (!fileLock.isValid()) {
+                checkState(!fileChannel.isOpen(), "invalid/unhandled state");
+                return;
+            }
+            fileLock.release();
+            fileChannel.close();
+            openedFile.close();
+        } finally {
+            closed = true;
+        }
+    }
+
+    public synchronized void position(final long newPosition) throws IOException {
+        checkNotClosed();
+        reopenFileChannelIfClosed();
+
+        fileChannel.position(newPosition);
+        currentPosition = newPosition;
+    }
+
+    private void checkNotClosed() {
+        checkState(!closed, "File has been closed");
+    }
+
+    private static void checkState(final boolean expression, final String errorMessage) {
+        if (!expression) {
+            throw new IllegalStateException(errorMessage);
+        }
+    }
+
+    public synchronized void force(final boolean metaData) throws IOException {
+        checkNotClosed();
+        reopenFileChannelIfClosed();
+
+        fileChannel.force(metaData);
+    }
+
+    public synchronized int write(final ByteBuffer src, final long position)
+            throws IOException {
+        checkNotClosed();
+        reopenFileChannelIfClosed();
+
+        return fileChannel.write(src, position);
+    }
+
+    public synchronized void read(final ByteBuffer buffer) throws IOException {
+        checkNotClosed();
+        reopenFileChannelIfClosed();
+
+        fileChannel.read(buffer);
+        currentPosition = fileChannel.position();
+    }
+
+    private void reopenFileChannelIfClosed() throws IOException {
+        if (!fileChannel.isOpen()) {
+            open();
+        }
+
+        if (fileChannel.position() != currentPosition) {
+            fileChannel.position(currentPosition);
+        }
+    }
+}

btm/src/main/java/bitronix/tm/journal/TransactionLogAppender.java

@@ -15,17 +15,9 @@
  */
 package bitronix.tm.journal;
 
-import bitronix.tm.utils.Uid;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.transaction.Status;
 import java.io.File;
 import java.io.IOException;
-import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.channels.FileLock;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
@@ -35,6 +27,13 @@
 import java.util.TreeSet;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import javax.transaction.Status;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import bitronix.tm.utils.Uid;
+
 /**
  * Used to write {@link TransactionLogRecord} objects to a log file.
  *
@@ -53,9 +52,7 @@ public class TransactionLogAppender {
     public static final int END_RECORD = 0x786e7442;
 
     private final File file;
-    private final RandomAccessFile randomeAccessFile;
-    private final FileChannel fc;
-    private final FileLock lock;
+    private final InterruptibleLockedRandomAccessFile randomAccessFile;
     private final TransactionLogHeader header;
 	private final long maxFileLength;
 	private final AtomicInteger outstandingWrites;
@@ -70,14 +67,10 @@ public class TransactionLogAppender {
      */
     public TransactionLogAppender(File file, long maxFileLength) throws IOException {
         this.file = file;
-        this.randomeAccessFile = new RandomAccessFile(file, "rw");
-        this.fc = randomeAccessFile.getChannel();
-        this.header = new TransactionLogHeader(fc, maxFileLength);
+        this.randomAccessFile = new InterruptibleLockedRandomAccessFile(file, "rw");
+        this.header = new TransactionLogHeader(randomAccessFile, maxFileLength);
         this.maxFileLength = maxFileLength;
-        this.lock = fc.tryLock(0, TransactionLogHeader.TIMESTAMP_HEADER, false);
-        if (this.lock == null)
-            throw new IOException("transaction log file " + file.getName() + " is locked. Is another instance already running?");
-
+        
         this.outstandingWrites = new AtomicInteger();
 
         this.danglingRecords = new HashMap<Uid, Set<String>>();
@@ -144,7 +137,7 @@ protected void writeLog(TransactionLogRecord tlog) throws IOException {
 
             final long writePosition = tlog.getWritePosition();
             while (buf.hasRemaining()) {
-            	fc.write(buf, writePosition + buf.position());
+            	randomAccessFile.write(buf, writePosition + buf.position());
             }
 
             trackOutstanding(status, gtrid, uniqueNames);
@@ -273,18 +266,14 @@ public long getPosition() {
     	return position;
     }
 
-
     /**
      * Close the appender and the underlying file.
      * @throws IOException if an I/O error occurs.
      */
     protected void close() throws IOException {
         header.setState(TransactionLogHeader.CLEAN_LOG_STATE);
-        fc.force(false);
-        if (lock != null)
-        	lock.release();
-        fc.close();
-        randomeAccessFile.close();
+        randomAccessFile.force(false);
+        randomAccessFile.close();
     }
 
     /**
@@ -304,7 +293,7 @@ protected TransactionLogCursor getCursor() throws IOException {
      */
     protected void force() throws IOException {
         if (log.isDebugEnabled()) { log.debug("forcing log writing"); }
-        fc.force(false);
+        randomAccessFile.force(false);
         if (log.isDebugEnabled()) { log.debug("done forcing log"); }
     }
 

btm/src/main/java/bitronix/tm/journal/TransactionLogHeader.java

@@ -69,7 +69,7 @@ public class TransactionLogHeader {
      */
     public final static byte UNCLEAN_LOG_STATE = -1;
 
-    private final FileChannel fc;
+    private final InterruptibleLockedRandomAccessFile file;
     private final long maxFileLength;
 
     private volatile int formatId;
@@ -79,25 +79,25 @@ public class TransactionLogHeader {
 
     /**
      * TransactionLogHeader are used to control headers of the specified RandomAccessFile.
-     * @param fc the file channel to read from.
+     * @param randomAccessFile the file to read from.
      * @param maxFileLength the max file length.
      * @throws IOException if an I/O error occurs.
      */
-    public TransactionLogHeader(FileChannel fc, long maxFileLength) throws IOException {
-        this.fc = fc;
+    public TransactionLogHeader(InterruptibleLockedRandomAccessFile randomAccessFile, long maxFileLength) throws IOException {
+        this.file = randomAccessFile;
         this.maxFileLength = maxFileLength;
 
-        fc.position(FORMAT_ID_HEADER);
+        randomAccessFile.position(FORMAT_ID_HEADER);
         ByteBuffer buf = ByteBuffer.allocate(4 + 8 + 1 + 8);
         while (buf.hasRemaining()) {
-            this.fc.read(buf);
+            this.file.read(buf);
         }
         buf.flip();
         formatId = buf.getInt();
         timestamp = buf.getLong();
         state = buf.get();
         position = buf.getLong();
-        fc.position(position);
+        randomAccessFile.position(position);
 
         if (log.isDebugEnabled()) { log.debug("read header " + this); }
     }
@@ -149,7 +149,7 @@ public void setFormatId(int formatId) throws IOException {
         buf.putInt(formatId);
         buf.flip();
         while (buf.hasRemaining()) {
-        	fc.write(buf, FORMAT_ID_HEADER + buf.position());
+        	file.write(buf, FORMAT_ID_HEADER + buf.position());
         }
         this.formatId = formatId;
     }
@@ -165,7 +165,7 @@ public void setTimestamp(long timestamp) throws IOException {
         buf.putLong(timestamp);
         buf.flip();
         while (buf.hasRemaining()) {
-        	fc.write(buf, TIMESTAMP_HEADER + buf.position());
+        	file.write(buf, TIMESTAMP_HEADER + buf.position());
         }
         this.timestamp = timestamp;
     }
@@ -181,7 +181,7 @@ public void setState(byte state) throws IOException {
         buf.put(state);
         buf.flip();
         while (buf.hasRemaining()) {
-        	fc.write(buf, STATE_HEADER + buf.position());
+        	file.write(buf, STATE_HEADER + buf.position());
         }
         this.state = state;
     }
@@ -202,11 +202,11 @@ public void setPosition(long position) throws IOException {
         buf.putLong(position);
         buf.flip();
         while (buf.hasRemaining()) {
-        	fc.write(buf, CURRENT_POSITION_HEADER + buf.position());
+        	file.write(buf, CURRENT_POSITION_HEADER + buf.position());
         }
 
         this.position = position;
-        fc.position(position);
+        file.position(position);
     }
 
     /**

btm/src/test/java/bitronix/tm/journal/ByteBufferUtil.java

@@ -0,0 +1,18 @@
+package bitronix.tm.journal;
+
+import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
+
+public final class ByteBufferUtil {
+
+    private ByteBufferUtil() {
+    }
+
+    public static ByteBuffer createByteBuffer(final String input)
+            throws UnsupportedEncodingException {
+        final byte[] inputArray = input.getBytes("UTF-8");
+        final ByteBuffer byteBuffer = ByteBuffer.wrap(inputArray);
+        return byteBuffer;
+    }
+
+}