WAL segments bigger than 2GB are allowed to be kept in WAL
@@ -67,6 +67,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Stream;
@@ -170,6 +171,7 @@ public final class CASDiskWriteAheadLog implements OWriteAheadLog {
private long segmentId = -1;
private final ScheduledFuture<?> recordsWriterFuture;
+ private final ReentrantLock recordsWriterLock = new ReentrantLock();
private final ConcurrentNavigableMap<OLogSequenceNumber, EventWrapper> events =
new ConcurrentSkipListMap<>();
@@ -1312,6 +1314,8 @@ public final class CASDiskWriteAheadLog implements OWriteAheadLog {
}
}
+ recordsWriterLock.lock();
+ try {
OWALRecord record = records.poll();
while (record != null) {
if (record instanceof WriteableWALRecord) {
@@ -1330,7 +1334,8 @@ public final class CASDiskWriteAheadLog implements OWriteAheadLog {
OLogManager.instance().errorNoDb(this, "WAL write was interrupted", e);
} catch (final ExecutionException e) {
OLogManager.instance().errorNoDb(this, "Error during writint of WAL data", e);
- throw OException.wrapException(new OStorageException("Error during writint of WAL data"), e);
+ throw OException.wrapException(
+ new OStorageException("Error during writint of WAL data"), e);
}
for (final OPair<Long, OWALFile> pair : fileCloseQueue) {
@@ -1364,6 +1369,9 @@ public final class CASDiskWriteAheadLog implements OWriteAheadLog {
writeBuffer = null;
writeBufferPageIndex = -1;
}
+ } finally {
+ recordsWriterLock.unlock();
+ }
}
public void addCheckpointListener(final OCheckpointRequestListener listener) {
@@ -1701,6 +1709,7 @@ public final class CASDiskWriteAheadLog implements OWriteAheadLog {
@Override
public void run() {
+ recordsWriterLock.lock();
try {
if (printPerformanceStatistic) {
printReport();
@@ -2010,6 +2019,8 @@ public final class CASDiskWriteAheadLog implements OWriteAheadLog {
} catch (final RuntimeException | Error e) {
OLogManager.instance().errorNoDb(this, "Error during WAL writing", e);
throw e;
+ } finally {
+ recordsWriterLock.unlock();
}
}
