Skip to content

Commit

Permalink
Implements interrupt-safe transaction log
Browse files Browse the repository at this point in the history
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 #45

Commit by Tibor Billes, Miklós Karakó, Balázs Póka
  • Loading branch information
billestibor authored and palacsint committed Jul 21, 2015
1 parent 1072c30 commit d13dc20
Show file tree
Hide file tree
Showing 11 changed files with 873 additions and 41 deletions.
20 changes: 19 additions & 1 deletion btm/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
<packaging>bundle</packaging>

<dependencies>
<dependency>
<dependency>
<groupId>javax.transaction</groupId>
<artifactId>jta</artifactId>
<scope>provided</scope>
Expand Down Expand Up @@ -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>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
Original file line number Diff line number Diff line change
@@ -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);
}
}
}
41 changes: 15 additions & 26 deletions btm/src/main/java/bitronix/tm/journal/TransactionLogAppender.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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.
*
Expand All @@ -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;
Expand All @@ -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>>();
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
}

/**
Expand All @@ -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"); }
}

Expand Down
24 changes: 12 additions & 12 deletions btm/src/main/java/bitronix/tm/journal/TransactionLogHeader.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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); }
}
Expand Down Expand Up @@ -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;
}
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand All @@ -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);
}

/**
Expand Down
18 changes: 18 additions & 0 deletions btm/src/test/java/bitronix/tm/journal/ByteBufferUtil.java
Original file line number Diff line number Diff line change
@@ -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;
}

}
Loading

0 comments on commit d13dc20

Please sign in to comment.