Implements interrupt-safe transaction log

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;
+    }
+
+}