ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dgovoruk...@apache.org
Subject [4/4] ignite git commit: IGNITE-9909 Merge FsyncWalManager and WalManager - Fixes #5013.
Date Fri, 09 Nov 2018 12:07:51 GMT
IGNITE-9909 Merge FsyncWalManager and WalManager - Fixes #5013.

Signed-off-by: Dmitriy Govorukhin <dmitriy.govorukhin@gmail.com>


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/889ce79b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/889ce79b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/889ce79b

Branch: refs/heads/master
Commit: 889ce79bba0187891d334fedbc63235ff85fabe6
Parents: cf81e5a
Author: Anton Kalashnikov <kaa.dev@yandex.ru>
Authored: Fri Nov 9 15:05:18 2018 +0300
Committer: Dmitriy Govorukhin <dmitriy.govorukhin@gmail.com>
Committed: Fri Nov 9 15:05:18 2018 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheProcessor.java    |   17 +-
 .../wal/FileWriteAheadLogManager.java           | 1017 +----
 .../wal/FsyncModeFileWriteAheadLogManager.java  | 3482 ------------------
 .../wal/filehandle/AbstractFileHandle.java      |   47 +
 .../wal/filehandle/FileHandleManager.java       |   81 +
 .../filehandle/FileHandleManagerFactory.java    |   90 +
 .../wal/filehandle/FileHandleManagerImpl.java   |  603 +++
 .../wal/filehandle/FileWriteHandle.java         |  113 +
 .../wal/filehandle/FileWriteHandleImpl.java     |  601 +++
 .../filehandle/FsyncFileHandleManagerImpl.java  |  157 +
 .../wal/filehandle/FsyncFileWriteHandle.java    |  845 +++++
 .../wal/serializer/RecordSerializerFactory.java |    2 +
 .../resources/META-INF/classnames.properties    |    5 -
 ...lFlushMultiNodeFailoverAbstractSelfTest.java |    9 +-
 .../wal/IgniteWalIteratorSwitchSegmentTest.java |   65 +-
 .../wal/memtracker/PageMemoryTracker.java       |   45 +-
 .../yardstick/IgniteAbstractBenchmark.java      |    7 +-
 17 files changed, 2615 insertions(+), 4571 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/889ce79b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 317378b..940e4a6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -62,7 +62,6 @@ import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.configuration.MemoryConfiguration;
 import org.apache.ignite.configuration.NearCacheConfiguration;
 import org.apache.ignite.configuration.TransactionConfiguration;
-import org.apache.ignite.configuration.WALMode;
 import org.apache.ignite.events.EventType;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
@@ -111,7 +110,6 @@ import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCa
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotDiscoveryMessage;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
 import org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager;
-import org.apache.ignite.internal.processors.cache.persistence.wal.FsyncModeFileWriteAheadLogManager;
 import org.apache.ignite.internal.processors.cache.query.GridCacheDistributedQueryManager;
 import org.apache.ignite.internal.processors.cache.query.GridCacheLocalQueryManager;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
@@ -214,9 +212,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
     private final boolean startClientCaches =
         IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_START_CACHES_ON_JOIN, false);
 
-    private final boolean walFsyncWithDedicatedWorker =
-        IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_WAL_FSYNC_WITH_DEDICATED_WORKER, false);
-
     /** Enables start caches in parallel. */
     private final boolean IGNITE_ALLOW_START_CACHES_IN_PARALLEL =
         IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_ALLOW_START_CACHES_IN_PARALLEL, true);
@@ -3041,13 +3036,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
 
             walMgr = ctx.plugins().createComponent(IgniteWriteAheadLogManager.class);
 
-            if (walMgr == null) {
-                if (ctx.config().getDataStorageConfiguration().getWalMode() == WALMode.FSYNC &&
-                    !walFsyncWithDedicatedWorker)
-                    walMgr = new FsyncModeFileWriteAheadLogManager(ctx);
-                else
-                    walMgr = new FileWriteAheadLogManager(ctx);
-            }
+            if (walMgr == null)
+                walMgr = new FileWriteAheadLogManager(ctx);
         }
         else {
             if (CU.isPersistenceEnabled(ctx.config()) && ctx.clientNode()) {
@@ -3531,7 +3521,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
      * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
      * @return Future that will be completed when cache is deployed.
      */
-    @SuppressWarnings("IfMayBeConditional")
     public IgniteInternalFuture<Boolean> dynamicStartCache(
         @Nullable CacheConfiguration ccfg,
         String cacheName,
@@ -3555,7 +3544,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
      *
      * @param ccfg Cache configuration.
      */
-    @SuppressWarnings("IfMayBeConditional")
     public IgniteInternalFuture<Boolean> dynamicStartSqlCache(
         CacheConfiguration ccfg
     ) {
@@ -3584,7 +3572,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
      * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
      * @return Future that will be completed when cache is deployed.
      */
-    @SuppressWarnings("IfMayBeConditional")
     public IgniteInternalFuture<Boolean> dynamicStartCache(
         @Nullable CacheConfiguration ccfg,
         String cacheName,

http://git-wip-us.apache.org/repos/asf/ignite/blob/889ce79b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
index fbc4f6c..0bd20bc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
@@ -26,12 +26,8 @@ import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
-import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
-import java.nio.MappedByteBuffer;
 import java.nio.channels.ClosedByInterruptException;
 import java.nio.file.FileAlreadyExistsException;
 import java.nio.file.Files;
@@ -46,17 +42,9 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.PriorityBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.LockSupport;
-import java.util.concurrent.locks.ReentrantLock;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
 import java.util.zip.ZipEntry;
@@ -79,7 +67,6 @@ import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
 import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
 import org.apache.ignite.internal.pagemem.wal.WALIterator;
 import org.apache.ignite.internal.pagemem.wal.WALPointer;
-import org.apache.ignite.internal.pagemem.wal.record.CheckpointRecord;
 import org.apache.ignite.internal.pagemem.wal.record.MarshalledRecord;
 import org.apache.ignite.internal.pagemem.wal.record.MemoryRecoveryRecord;
 import org.apache.ignite.internal.pagemem.wal.record.PageSnapshot;
@@ -101,6 +88,10 @@ import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolde
 import org.apache.ignite.internal.processors.cache.persistence.wal.aware.SegmentAware;
 import org.apache.ignite.internal.processors.cache.persistence.wal.crc.FastCrc;
 import org.apache.ignite.internal.processors.cache.persistence.wal.crc.IgniteDataIntegrityViolationException;
+import org.apache.ignite.internal.processors.cache.persistence.wal.filehandle.AbstractFileHandle;
+import org.apache.ignite.internal.processors.cache.persistence.wal.filehandle.FileHandleManagerFactory;
+import org.apache.ignite.internal.processors.cache.persistence.wal.filehandle.FileHandleManager;
+import org.apache.ignite.internal.processors.cache.persistence.wal.filehandle.FileWriteHandle;
 import org.apache.ignite.internal.processors.cache.persistence.wal.io.FileInput;
 import org.apache.ignite.internal.processors.cache.persistence.wal.io.LockedSegmentFileInputFactory;
 import org.apache.ignite.internal.processors.cache.persistence.wal.io.SegmentFileInputFactory;
@@ -114,7 +105,6 @@ import org.apache.ignite.internal.processors.cache.persistence.wal.serializer.Re
 import org.apache.ignite.internal.processors.failure.FailureProcessor;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
-import org.apache.ignite.internal.util.GridUnsafe;
 import org.apache.ignite.internal.util.future.GridFinishedFuture;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
 import org.apache.ignite.internal.util.typedef.CI1;
@@ -139,48 +129,20 @@ import static org.apache.ignite.IgniteSystemProperties.IGNITE_CHECKPOINT_TRIGGER
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_THRESHOLD_WAL_ARCHIVE_SIZE_PERCENTAGE;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAL_COMPRESSOR_WORKER_THREAD_CNT;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAL_MMAP;
-import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAL_SEGMENT_SYNC_TIMEOUT;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAL_SERIALIZER_VERSION;
-import static org.apache.ignite.configuration.WALMode.LOG_ONLY;
 import static org.apache.ignite.events.EventType.EVT_WAL_SEGMENT_ARCHIVED;
 import static org.apache.ignite.events.EventType.EVT_WAL_SEGMENT_COMPACTED;
 import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
 import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
-import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.SWITCH_SEGMENT_RECORD;
-import static org.apache.ignite.internal.processors.cache.persistence.wal.SegmentedRingByteBuffer.BufferMode.DIRECT;
+import static org.apache.ignite.internal.processors.cache.persistence.wal.serializer.RecordSerializerFactory.LATEST_SERIALIZER_VERSION;
 import static org.apache.ignite.internal.processors.cache.persistence.wal.serializer.RecordV1Serializer.HEADER_RECORD_SIZE;
 import static org.apache.ignite.internal.processors.cache.persistence.wal.serializer.RecordV1Serializer.readSegmentHeader;
-import static org.apache.ignite.internal.util.IgniteUtils.findField;
-import static org.apache.ignite.internal.util.IgniteUtils.findNonPublicMethod;
-import static org.apache.ignite.internal.util.IgniteUtils.sleep;
 
 /**
  * File WAL manager.
  */
 @SuppressWarnings("IfMayBeConditional")
 public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter implements IgniteWriteAheadLogManager {
-    /** Dfault wal segment sync timeout. */
-    public static final long DFLT_WAL_SEGMENT_SYNC_TIMEOUT = 500L;
-    /** {@link MappedByteBuffer#force0(java.io.FileDescriptor, long, long)}. */
-    private static final Method force0 = findNonPublicMethod(
-        MappedByteBuffer.class, "force0",
-        java.io.FileDescriptor.class, long.class, long.class
-    );
-
-    /** {@link MappedByteBuffer#mappingOffset()}. */
-    private static final Method mappingOffset = findNonPublicMethod(MappedByteBuffer.class, "mappingOffset");
-
-    /** {@link MappedByteBuffer#mappingAddress(long)}. */
-    private static final Method mappingAddress = findNonPublicMethod(
-        MappedByteBuffer.class, "mappingAddress", long.class
-    );
-
-    /** {@link MappedByteBuffer#fd} */
-    private static final Field fd = findField(MappedByteBuffer.class, "fd");
-
-    /** Page size. */
-    private static final int PAGE_SIZE = GridUnsafe.pageSize();
-
     /** */
     private static final FileDescriptor[] EMPTY_DESCRIPTORS = new FileDescriptor[0];
 
@@ -235,19 +197,12 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
         }
     };
 
-    /** Latest serializer version to use. */
-    private static final int LATEST_SERIALIZER_VERSION = 2;
-
     /** Buffer size. */
     private static final int BUF_SIZE = 1024 * 1024;
 
     /** Use mapped byte buffer. */
     private final boolean mmap = IgniteSystemProperties.getBoolean(IGNITE_WAL_MMAP, true);
 
-    /** {@link FileWriteHandle#written} atomic field updater. */
-    private static final AtomicLongFieldUpdater<FileWriteHandle> WRITTEN_UPD =
-        AtomicLongFieldUpdater.newUpdater(FileWriteHandle.class, "written");
-
     /**
      * Percentage of archive size for checkpoint trigger. Need for calculate max size of WAL after last checkpoint.
      * Checkpoint should be triggered when max size of WAL after last checkpoint more than maxWallArchiveSize * thisValue
@@ -288,9 +243,6 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
     /** WAL flush frequency. Makes sense only for {@link WALMode#BACKGROUND} log WALMode. */
     private final long flushFreq;
 
-    /** Fsync delay. */
-    private final long fsyncDelay;
-
     /** */
     private final DataStorageConfiguration dsCfg;
 
@@ -347,9 +299,12 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
     /** */
     private final ThreadLocal<WALPointer> lastWALPtr = new ThreadLocal<>();
 
-    /** Current log segment handle */
+    /** Current log segment handle. */
     private volatile FileWriteHandle currHnd;
 
+    /** File handle manager. */
+    private FileHandleManager fileHandleManager;
+
     /** */
     private volatile WALDisableContext walDisableContext;
 
@@ -378,17 +333,11 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
      */
     @Nullable private volatile GridTimeoutObject nextAutoArchiveTimeoutObj;
 
-    /** WAL writer worker. */
-    private WALWriter walWriter;
-
     /**
      * Listener invoked for each segment file IO initializer.
      */
     @Nullable private volatile IgniteInClosure<FileIO> createWalFileListener;
 
-    /** Wal segment sync worker. */
-    private WalSegmentSyncer walSegmentSyncWorker;
-
     /**
      * Manage of segment location.
      */
@@ -397,6 +346,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
     /** Segment factory with ability locked segment during reading. */
     private SegmentFileInputFactory lockedSegmentFileInputFactory;
 
+    private FileHandleManagerFactory fileHandleManagerFactory;
+
     /**
      * @param ctx Kernal context.
      */
@@ -412,9 +363,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
         maxWalSegmentSize = dsCfg.getWalSegmentSize();
         mode = dsCfg.getWalMode();
         flushFreq = dsCfg.getWalFlushFrequency();
-        fsyncDelay = dsCfg.getWalFsyncDelayNanos();
         alwaysWriteFullPages = dsCfg.isAlwaysWriteFullPages();
-        ioFactory = new RandomAccessFileIOFactory();
+        ioFactory = mode == WALMode.FSYNC ? dsCfg.getFileIOFactory() : new RandomAccessFileIOFactory();
         segmentFileInputFactory = new SimpleSegmentFileInputFactory();
         walAutoArchiveAfterInactivity = dsCfg.getWalAutoArchiveAfterInactivity();
 
@@ -422,6 +372,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
 
         evt = ctx.event();
         failureProcessor = ctx.failure();
+
+        fileHandleManagerFactory = new FileHandleManagerFactory(dsCfg);
     }
 
     /**
@@ -512,16 +464,11 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
 
             walDisableContext = cctx.walState().walDisableContext();
 
-            if (mode != WALMode.NONE && mode != WALMode.FSYNC) {
-                walSegmentSyncWorker = new WalSegmentSyncer(igCfg.getIgniteInstanceName(),
-                    cctx.kernalContext().log(WalSegmentSyncer.class));
+            fileHandleManager = fileHandleManagerFactory.build(
+                cctx, metrics, mmap, lastWALPtr::get, serializer, this::currentHandle
+            );
 
-                if (log.isInfoEnabled())
-                    log.info("Started write-ahead log manager [mode=" + mode + ']');
-            }
-            else
-                U.quietAndWarn(log, "Started write-ahead log manager in NONE mode, persisted data may be lost in " +
-                    "a case of unexpected node failure. Make sure to deactivate the cluster before shutdown.");
+            fileHandleManager.start();
 
             lockedSegmentFileInputFactory = new LockedSegmentFileInputFactory(
                 segmentAware,
@@ -541,8 +488,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
             new IgniteThread(archiver).start();
         }
 
-        if (walSegmentSyncWorker != null)
-            new IgniteThread(walSegmentSyncWorker).start();
+        fileHandleManager.onActivate();
 
         if (compressor != null)
             compressor.start();
@@ -625,22 +571,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
         if (timeoutObj != null)
             cctx.time().removeTimeoutObject(timeoutObj);
 
-        final FileWriteHandle currHnd = currentHandle();
-
         try {
-            if (mode == WALMode.BACKGROUND) {
-                if (currHnd != null)
-                    currHnd.flush(null);
-            }
-
-            if (currHnd != null)
-                currHnd.close(false);
-
-            if (walSegmentSyncWorker != null)
-                walSegmentSyncWorker.shutdown();
-
-            if (walWriter != null)
-                walWriter.shutdown();
+            fileHandleManager.onDeactivate();
 
             segmentAware.interrupt();
 
@@ -660,7 +592,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
             }
         }
         catch (Exception e) {
-            U.error(log, "Failed to gracefully close WAL segment: " + this.currHnd.fileIO, e);
+            U.error(log, "Failed to gracefully close WAL segment: " + this.currHnd, e);
         }
     }
 
@@ -706,10 +638,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
 
         FileWALPointer filePtr = (FileWALPointer)lastPtr;
 
-        walWriter = new WALWriter(log);
-
-        if (!mmap)
-            new IgniteThread(walWriter).start();
+        fileHandleManager.resumeLogging();
 
         currHnd = restoreWriteHandle(filePtr);
 
@@ -717,16 +646,16 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
         if (filePtr == null)
             currHnd.writeHeader();
 
-        if (currHnd.serializer.version() != serializer.version()) {
+        if (currHnd.serializerVersion() != serializer.version()) {
             if (log.isInfoEnabled())
                 log.info("Record serializer version change detected, will start logging with a new WAL record " +
                     "serializer to a new WAL segment [curFile=" + currHnd + ", newVer=" + serializer.version() +
-                    ", oldVer=" + currHnd.serializer.version() + ']');
+                    ", oldVer=" + currHnd.serializerVersion() + ']');
 
             rollOver(currHnd, null);
         }
 
-        currHnd.resume = false;
+        currHnd.finishResumeLogging();
 
         if (mode == WALMode.BACKGROUND) {
             backgroundFlushSchedule = cctx.time().schedule(new Runnable() {
@@ -803,9 +732,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
         final FileWriteHandle handle = currentHandle();
 
         try {
-            handle.buf.close();
-
-            rollOver(handle, null);
+            closeBufAndRollover(handle, null, RolloverType.NONE);
         }
         catch (IgniteCheckedException e) {
             U.error(log, "Unable to perform segment rollover: " + e.getMessage(), e);
@@ -895,7 +822,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
     ) throws IgniteCheckedException {
         long idx = currWriteHandle.getSegmentId();
 
-        currWriteHandle.buf.close();
+        currWriteHandle.closeBuffer();
 
         FileWriteHandle res = rollOver(currWriteHandle, rolloverType == RolloverType.NEXT_SEGMENT ? rec : null);
 
@@ -907,28 +834,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
 
     /** {@inheritDoc} */
     @Override public void flush(WALPointer ptr, boolean explicitFsync) throws IgniteCheckedException, StorageException {
-        if (serializer == null || mode == WALMode.NONE)
-            return;
-
-        FileWriteHandle cur = currentHandle();
-
-        // WAL manager was not started (client node).
-        if (cur == null)
-            return;
-
-        FileWALPointer filePtr = (FileWALPointer)(ptr == null ? lastWALPtr.get() : ptr);
-
-        if (mode == LOG_ONLY)
-            cur.flushOrWait(filePtr);
-
-        if (!explicitFsync && mode != WALMode.FSYNC)
-            return; // No need to sync in LOG_ONLY or BACKGROUND unless explicit fsync is required.
-
-        // No need to sync if was rolled over.
-        if (filePtr != null && !cur.needFsync(filePtr))
-            return;
-
-        cur.fsync(filePtr);
+        fileHandleManager.flush(ptr, explicitFsync);
     }
 
     /** {@inheritDoc} */
@@ -1330,25 +1236,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
                     log.info("Resuming logging to WAL segment [file=" + curFile.getAbsolutePath() +
                         ", offset=" + off + ", ver=" + serVer + ']');
 
-                SegmentedRingByteBuffer rbuf;
-
-                if (mmap) {
-                    MappedByteBuffer buf = fileIO.map((int)maxWalSegmentSize);
-
-                    rbuf = new SegmentedRingByteBuffer(buf, metrics);
-                }
-                else
-                    rbuf = new SegmentedRingByteBuffer(dsCfg.getWalBufferSize(), maxWalSegmentSize, DIRECT, metrics);
-
-                if (lastReadPtr != null)
-                    rbuf.init(lastReadPtr.fileOffset() + lastReadPtr.length());
-
-                FileWriteHandle hnd = new FileWriteHandle(
-                    fileIO,
-                    off + len,
-                    true,
-                    ser,
-                    rbuf);
+                FileWriteHandle hnd = fileHandleManager.initHandle(fileIO, off + len, ser);
 
                 if (archiver0 != null)
                     segmentAware.curAbsWalIdx(absIdx);
@@ -1393,8 +1281,6 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
             if (log.isDebugEnabled())
                 log.debug("Switching to a new WAL segment: " + nextFile.getAbsolutePath());
 
-            SegmentedRingByteBuffer rbuf = null;
-
             SegmentIO fileIO = null;
 
             FileWriteHandle hnd;
@@ -1409,20 +1295,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
                     if (lsnr != null)
                         lsnr.apply(fileIO);
 
-                    if (mmap) {
-                        MappedByteBuffer buf = fileIO.map((int)maxWalSegmentSize);
 
-                        rbuf = new SegmentedRingByteBuffer(buf, metrics);
-                    }
-                    else
-                        rbuf = cur.buf.reset();
-
-                    hnd = new FileWriteHandle(
-                        fileIO,
-                        0,
-                        false,
-                        serializer,
-                        rbuf);
+                    hnd = fileHandleManager.nextHandle(fileIO, serializer);
 
                     if (interrupted)
                         Thread.currentThread().interrupt();
@@ -1444,12 +1318,6 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
 
                         fileIO = null;
                     }
-
-                    if (rbuf != null) {
-                        rbuf.free();
-
-                        rbuf = null;
-                    }
                 }
             }
 
@@ -2416,7 +2284,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
      * @param ver Version.
      * @param compacted Compacted flag.
      */
-    @NotNull private static ByteBuffer prepareSerializerVersionBuffer(long idx, int ver, boolean compacted, ByteBuffer buf) {
+    @NotNull public static ByteBuffer prepareSerializerVersionBuffer(long idx, int ver, boolean compacted, ByteBuffer buf) {
         // Write record type.
         buf.put((byte) (WALRecord.RecordType.HEADER_RECORD.ordinal() + 1));
 
@@ -2452,33 +2320,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
     /**
      *
      */
-    private abstract static class FileHandle {
-        /** I/O interface for read/write operations with file */
-        SegmentIO fileIO;
-
-        /** Segment idx corresponded to fileIo*/
-        final long segmentIdx;
-
-        /**
-         * @param fileIO I/O interface for read/write operations of FileHandle.
-         */
-        private FileHandle(@NotNull SegmentIO fileIO) {
-            this.fileIO = fileIO;
-            segmentIdx = fileIO.getSegmentId();
-        }
-
-        /**
-         * @return Absolute WAL segment file index (incremental counter).
-         */
-        public long getSegmentId(){
-            return segmentIdx;
-        }
-    }
-
-    /**
-     *
-     */
-    public static class ReadFileHandle extends FileHandle implements AbstractWalRecordsIterator.AbstractReadFileHandle {
+    public static class ReadFileHandle extends AbstractFileHandle implements AbstractWalRecordsIterator.AbstractReadFileHandle {
         /** Entry serializer. */
         RecordSerializer ser;
 
@@ -2489,7 +2331,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
         private final SegmentAware segmentAware;
 
         /**
-         * @param fileIO I/O interface for read/write operations of FileHandle.
+         * @param fileIO I/O interface for read/write operations of AbstractFileHandle.
          * @param ser Entry serializer.
          * @param in File input.
          * @param aware Segment aware.
@@ -2542,459 +2384,6 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
     }
 
     /**
-     * File handle for one log segment.
-     */
-    @SuppressWarnings("SignalWithoutCorrespondingAwait")
-    private class FileWriteHandle extends FileHandle {
-        /** */
-        private final RecordSerializer serializer;
-
-        /** Created on resume logging. */
-        private volatile boolean resume;
-
-        /**
-         * Position in current file after the end of last written record (incremented after file channel write
-         * operation)
-         */
-        volatile long written;
-
-        /** */
-        private volatile long lastFsyncPos;
-
-        /** Stop guard to provide warranty that only one thread will be successful in calling {@link #close(boolean)} */
-        private final AtomicBoolean stop = new AtomicBoolean(false);
-
-        /** */
-        private final Lock lock = new ReentrantLock();
-
-        /** Condition for timed wait of several threads, see {@link DataStorageConfiguration#getWalFsyncDelayNanos()} */
-        private final Condition fsync = lock.newCondition();
-
-        /**
-         * Next segment available condition. Protection from "spurious wakeup" is provided by predicate {@link
-         * #fileIO}=<code>null</code>
-         */
-        private final Condition nextSegment = lock.newCondition();
-
-        /** Buffer. */
-        private final SegmentedRingByteBuffer buf;
-
-        /**
-         * @param fileIO I/O file interface to use
-         * @param pos Position.
-         * @param resume Created on resume logging flag.
-         * @param serializer Serializer.
-         * @param buf Buffer.
-         * @throws IOException If failed.
-         */
-        private FileWriteHandle(
-            SegmentIO fileIO,
-            long pos,
-            boolean resume,
-            RecordSerializer serializer,
-            SegmentedRingByteBuffer buf
-        ) throws IOException {
-            super(fileIO);
-
-            assert serializer != null;
-
-            if (!mmap)
-                fileIO.position(pos);
-
-            this.serializer = serializer;
-
-            written = pos;
-            lastFsyncPos = pos;
-            this.resume = resume;
-            this.buf = buf;
-        }
-
-        /**
-         * Write serializer version to current handle.
-         */
-        public void writeHeader() {
-            SegmentedRingByteBuffer.WriteSegment seg = buf.offer(HEADER_RECORD_SIZE);
-
-            assert seg != null && seg.position() > 0;
-
-            prepareSerializerVersionBuffer(getSegmentId(), serializerVersion(), false, seg.buffer());
-
-            seg.release();
-        }
-
-        /**
-         * @param rec Record to be added to write queue.
-         * @return Pointer or null if roll over to next segment is required or already started by other thread.
-         * @throws StorageException If failed.
-         * @throws IgniteCheckedException If failed.
-         */
-        @Nullable private WALPointer addRecord(WALRecord rec) throws StorageException, IgniteCheckedException {
-            assert rec.size() > 0 : rec;
-
-            for (;;) {
-                checkNode();
-
-                SegmentedRingByteBuffer.WriteSegment seg;
-
-                // Buffer can be in open state in case of resuming with different serializer version.
-                if (rec.type() == SWITCH_SEGMENT_RECORD && !currHnd.resume)
-                    seg = buf.offerSafe(rec.size());
-                else
-                    seg = buf.offer(rec.size());
-
-                FileWALPointer ptr = null;
-
-                if (seg != null) {
-                    try {
-                        int pos = (int)(seg.position() - rec.size());
-
-                        ByteBuffer buf = seg.buffer();
-
-                        if (buf == null)
-                            return null; // Can not write to this segment, need to switch to the next one.
-
-                        ptr = new FileWALPointer(getSegmentId(), pos, rec.size());
-
-                        rec.position(ptr);
-
-                        fillBuffer(buf, rec);
-
-                        if (mmap) {
-                            // written field must grow only, but segment with greater position can be serialized
-                            // earlier than segment with smaller position.
-                            while (true) {
-                                long written0 = written;
-
-                                if (seg.position() > written0) {
-                                    if (WRITTEN_UPD.compareAndSet(this, written0, seg.position()))
-                                        break;
-                                }
-                                else
-                                    break;
-                            }
-                        }
-
-                        return ptr;
-                    }
-                    finally {
-                        seg.release();
-
-                        if (mode == WALMode.BACKGROUND && rec instanceof CheckpointRecord)
-                            flushOrWait(ptr);
-                    }
-                }
-                else
-                    walWriter.flushAll();
-            }
-        }
-
-        /**
-         * Flush or wait for concurrent flush completion.
-         *
-         * @param ptr Pointer.
-         */
-        private void flushOrWait(FileWALPointer ptr) throws IgniteCheckedException {
-            if (ptr != null) {
-                // If requested obsolete file index, it must be already flushed by close.
-                if (ptr.index() != getSegmentId())
-                    return;
-            }
-
-            flush(ptr);
-        }
-
-        /**
-         * @param ptr Pointer.
-         */
-        private void flush(FileWALPointer ptr) throws IgniteCheckedException {
-            if (ptr == null) { // Unconditional flush.
-                walWriter.flushAll();
-
-                return;
-            }
-
-            assert ptr.index() == getSegmentId();
-
-            walWriter.flushBuffer(ptr.fileOffset());
-        }
-
-        /**
-         * @param buf Buffer.
-         * @param rec WAL record.
-         * @throws IgniteCheckedException If failed.
-         */
-        private void fillBuffer(ByteBuffer buf, WALRecord rec) throws IgniteCheckedException {
-            try {
-                serializer.writeRecord(rec, buf);
-            }
-            catch (RuntimeException e) {
-                throw new IllegalStateException("Failed to write record: " + rec, e);
-            }
-        }
-
-        /**
-         * Non-blocking check if this pointer needs to be sync'ed.
-         *
-         * @param ptr WAL pointer to check.
-         * @return {@code False} if this pointer has been already sync'ed.
-         */
-        private boolean needFsync(FileWALPointer ptr) {
-            // If index has changed, it means that the log was rolled over and already sync'ed.
-            // If requested position is smaller than last sync'ed, it also means all is good.
-            // If position is equal, then our record is the last not synced.
-            return getSegmentId() == ptr.index() && lastFsyncPos <= ptr.fileOffset();
-        }
-
-        /**
-         * @return Pointer to the end of the last written record (probably not fsync-ed).
-         */
-        private FileWALPointer position() {
-            lock.lock();
-
-            try {
-                return new FileWALPointer(getSegmentId(), (int)written, 0);
-            }
-            finally {
-                lock.unlock();
-            }
-        }
-
-        /**
-         * @param ptr Pointer to sync.
-         * @throws StorageException If failed.
-         */
-        private void fsync(FileWALPointer ptr) throws StorageException, IgniteCheckedException {
-            lock.lock();
-
-            try {
-                if (ptr != null) {
-                    if (!needFsync(ptr))
-                        return;
-
-                    if (fsyncDelay > 0 && !stop.get()) {
-                        // Delay fsync to collect as many updates as possible: trade latency for throughput.
-                        U.await(fsync, fsyncDelay, TimeUnit.NANOSECONDS);
-
-                        if (!needFsync(ptr))
-                            return;
-                    }
-                }
-
-                flushOrWait(ptr);
-
-                if (stop.get())
-                    return;
-
-                long lastFsyncPos0 = lastFsyncPos;
-                long written0 = written;
-
-                if (lastFsyncPos0 != written0) {
-                    // Fsync position must be behind.
-                    assert lastFsyncPos0 < written0 : "lastFsyncPos=" + lastFsyncPos0 + ", written=" + written0;
-
-                    boolean metricsEnabled = metrics.metricsEnabled();
-
-                    long start = metricsEnabled ? System.nanoTime() : 0;
-
-                    if (mmap) {
-                        long pos = ptr == null ? -1 : ptr.fileOffset();
-
-                        List<SegmentedRingByteBuffer.ReadSegment> segs = buf.poll(pos);
-
-                        if (segs != null) {
-                            assert segs.size() == 1;
-
-                            SegmentedRingByteBuffer.ReadSegment seg = segs.get(0);
-
-                            int off = seg.buffer().position();
-                            int len = seg.buffer().limit() - off;
-
-                            fsync((MappedByteBuffer)buf.buf, off, len);
-
-                            seg.release();
-                        }
-                    }
-                    else
-                        walWriter.force();
-
-                    lastFsyncPos = written;
-
-                    if (fsyncDelay > 0)
-                        fsync.signalAll();
-
-                    long end = metricsEnabled ? System.nanoTime() : 0;
-
-                    if (metricsEnabled)
-                        metrics.onFsync(end - start);
-                }
-            }
-            finally {
-                lock.unlock();
-            }
-        }
-
-        /**
-         * @param buf Mapped byte buffer..
-         * @param off Offset.
-         * @param len Length.
-         */
-        private void fsync(MappedByteBuffer buf, int off, int len) throws IgniteCheckedException {
-            try {
-                long mappedOff = (Long)mappingOffset.invoke(buf);
-
-                assert mappedOff == 0 : mappedOff;
-
-                long addr = (Long)mappingAddress.invoke(buf, mappedOff);
-
-                long delta = (addr + off) % PAGE_SIZE;
-
-                long alignedAddr = (addr + off) - delta;
-
-                force0.invoke(buf, fd.get(buf), alignedAddr, len + delta);
-            }
-            catch (IllegalAccessException | InvocationTargetException e) {
-                throw new IgniteCheckedException(e);
-            }
-        }
-
-        /**
-         * @return {@code true} If this thread actually closed the segment.
-         * @throws IgniteCheckedException If failed.
-         * @throws StorageException If failed.
-         */
-        private boolean close(boolean rollOver) throws IgniteCheckedException, StorageException {
-            if (stop.compareAndSet(false, true)) {
-                lock.lock();
-
-                try {
-                    flushOrWait(null);
-
-                    try {
-                        RecordSerializer backwardSerializer = new RecordSerializerFactoryImpl(cctx)
-                            .createSerializer(serializerVer);
-
-                        SwitchSegmentRecord segmentRecord = new SwitchSegmentRecord();
-
-                        int switchSegmentRecSize = backwardSerializer.size(segmentRecord);
-
-                        if (rollOver && written < (maxWalSegmentSize - switchSegmentRecSize)) {
-                            segmentRecord.size(switchSegmentRecSize);
-
-                            WALPointer segRecPtr = addRecord(segmentRecord);
-
-                            if (segRecPtr != null)
-                                fsync((FileWALPointer)segRecPtr);
-                        }
-
-                        if (mmap) {
-                            List<SegmentedRingByteBuffer.ReadSegment> segs = buf.poll(maxWalSegmentSize);
-
-                            if (segs != null) {
-                                assert segs.size() == 1;
-
-                                segs.get(0).release();
-                            }
-                        }
-
-                        // Do the final fsync.
-                        if (mode != WALMode.NONE) {
-                            if (mmap)
-                                ((MappedByteBuffer)buf.buf).force();
-                            else
-                                fileIO.force();
-
-                            lastFsyncPos = written;
-                        }
-
-                        if (mmap) {
-                            try {
-                                fileIO.close();
-                            }
-                            catch (IOException ignore) {
-                                // No-op.
-                            }
-                        }
-                        else {
-                            walWriter.close();
-
-                            if (!rollOver)
-                                buf.free();
-                        }
-                    }
-                    catch (IOException e) {
-                        throw new StorageException("Failed to close WAL write handle [idx=" + getSegmentId() + "]", e);
-                    }
-
-                    if (log.isDebugEnabled())
-                        log.debug("Closed WAL write handle [idx=" + getSegmentId() + "]");
-
-                    return true;
-                }
-                finally {
-                    if (mmap)
-                        buf.free();
-
-                    lock.unlock();
-                }
-            }
-            else
-                return false;
-        }
-
-        /**
-         * Signals next segment available to wake up other worker threads waiting for WAL to write
-         */
-        private void signalNextAvailable() {
-            lock.lock();
-
-            try {
-                assert cctx.kernalContext().invalid() ||
-                    written == lastFsyncPos || mode != WALMode.FSYNC :
-                    "fsync [written=" + written + ", lastFsync=" + lastFsyncPos + ", idx=" + getSegmentId() + ']';
-
-                fileIO = null;
-
-                nextSegment.signalAll();
-            }
-            finally {
-                lock.unlock();
-            }
-        }
-
-        /**
-         *
-         */
-        private void awaitNext() {
-            lock.lock();
-
-            try {
-                while (fileIO != null)
-                    U.awaitQuiet(nextSegment);
-            }
-            finally {
-                lock.unlock();
-            }
-        }
-
-        /**
-         * @return Safely reads current position of the file channel as String. Will return "null" if channel is null.
-         */
-        private String safePosition() {
-            FileIO io = fileIO;
-
-            if (io == null)
-                return "null";
-
-            try {
-                return String.valueOf(io.position());
-            }
-            catch (IOException e) {
-                return "{Failed to read channel position: " + e.getMessage() + '}';
-            }
-        }
-    }
-
-    /**
      * Iterator over WAL-log.
      */
     private static class RecordsIterator extends AbstractWalRecordsIterator {
@@ -3282,7 +2671,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
         FileWriteHandle hnd = currentHandle();
 
         try {
-            hnd.flush(null);
+            hnd.flushAll();
         }
         catch (Exception e) {
             U.warn(log, "Failed to flush WAL record queue", e);
@@ -3290,344 +2679,6 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
     }
 
     /**
-     * WAL writer worker.
-     */
-    private class WALWriter extends GridWorker {
-        /** Unconditional flush. */
-        private static final long UNCONDITIONAL_FLUSH = -1L;
-
-        /** File close. */
-        private static final long FILE_CLOSE = -2L;
-
-        /** File force. */
-        private static final long FILE_FORCE = -3L;
-
-        /** Err. */
-        private volatile Throwable err;
-
-        //TODO: replace with GC free data structure.
-        /** Parked threads. */
-        final Map<Thread, Long> waiters = new ConcurrentHashMap<>();
-
-        /**
-         * Default constructor.
-         *
-         * @param log Logger.
-         */
-        WALWriter(IgniteLogger log) {
-            super(cctx.igniteInstanceName(), "wal-write-worker%" + cctx.igniteInstanceName(), log,
-                cctx.kernalContext().workersRegistry());
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void body() {
-            Throwable err = null;
-
-            try {
-                while (!isCancelled()) {
-                    onIdle();
-
-                    while (waiters.isEmpty()) {
-                        if (!isCancelled()) {
-                            blockingSectionBegin();
-
-                            try {
-                                LockSupport.park();
-                            }
-                            finally {
-                                blockingSectionEnd();
-                            }
-                        }
-                        else {
-                            unparkWaiters(Long.MAX_VALUE);
-
-                            return;
-                        }
-                    }
-
-                    Long pos = null;
-
-                    for (Long val : waiters.values()) {
-                        if (val > Long.MIN_VALUE)
-                            pos = val;
-                    }
-
-                    updateHeartbeat();
-
-                    if (pos == null)
-                        continue;
-                    else if (pos < UNCONDITIONAL_FLUSH) {
-                        try {
-                            assert pos == FILE_CLOSE || pos == FILE_FORCE : pos;
-
-                            if (pos == FILE_CLOSE)
-                                currHnd.fileIO.close();
-                            else if (pos == FILE_FORCE)
-                                currHnd.fileIO.force();
-                        }
-                        catch (IOException e) {
-                            log.error("Exception in WAL writer thread: ", e);
-
-                            err = e;
-
-                            unparkWaiters(Long.MAX_VALUE);
-
-                            return;
-                        }
-
-                        unparkWaiters(pos);
-                    }
-
-                    updateHeartbeat();
-
-                    List<SegmentedRingByteBuffer.ReadSegment> segs = currentHandle().buf.poll(pos);
-
-                    if (segs == null) {
-                        unparkWaiters(pos);
-
-                        continue;
-                    }
-
-                    for (int i = 0; i < segs.size(); i++) {
-                        SegmentedRingByteBuffer.ReadSegment seg = segs.get(i);
-
-                        updateHeartbeat();
-
-                        try {
-                            writeBuffer(seg.position(), seg.buffer());
-                        }
-                        catch (Throwable e) {
-                            log.error("Exception in WAL writer thread:", e);
-
-                            err = e;
-                        }
-                        finally {
-                            seg.release();
-
-                            long p = pos <= UNCONDITIONAL_FLUSH || err != null ? Long.MAX_VALUE : currentHandle().written;
-
-                            unparkWaiters(p);
-                        }
-                    }
-                }
-            }
-            catch (Throwable t) {
-                err = t;
-            }
-            finally {
-                unparkWaiters(Long.MAX_VALUE);
-
-                if (err == null && !isCancelled)
-                    err = new IllegalStateException("Worker " + name() + " is terminated unexpectedly");
-
-                if (err instanceof OutOfMemoryError)
-                    cctx.kernalContext().failure().process(new FailureContext(CRITICAL_ERROR, err));
-                else if (err != null)
-                    cctx.kernalContext().failure().process(new FailureContext(SYSTEM_WORKER_TERMINATION, err));
-            }
-        }
-
-        /**
-         * Shutdowns thread.
-         */
-        public void shutdown() throws IgniteInterruptedCheckedException {
-            U.cancel(this);
-
-            LockSupport.unpark(runner());
-
-            U.join(runner());
-        }
-
-        /**
-         * Unparks waiting threads.
-         *
-         * @param pos Pos.
-         */
-        private void unparkWaiters(long pos) {
-            assert pos > Long.MIN_VALUE : pos;
-
-            for (Map.Entry<Thread, Long> e : waiters.entrySet()) {
-                Long val = e.getValue();
-
-                if (val <= pos) {
-                    if (val != Long.MIN_VALUE)
-                        waiters.put(e.getKey(), Long.MIN_VALUE);
-
-                    LockSupport.unpark(e.getKey());
-                }
-            }
-        }
-
-        /**
-         * Forces all made changes to the file.
-         */
-        void force() throws IgniteCheckedException {
-            flushBuffer(FILE_FORCE);
-        }
-
-        /**
-         * Closes file.
-         */
-        void close() throws IgniteCheckedException {
-            flushBuffer(FILE_CLOSE);
-        }
-
-        /**
-         * Flushes all data from the buffer.
-         */
-        void flushAll() throws IgniteCheckedException {
-            flushBuffer(UNCONDITIONAL_FLUSH);
-        }
-
-        /**
-         * @param expPos Expected position.
-         */
-        void flushBuffer(long expPos) throws IgniteCheckedException {
-            if (mmap)
-                return;
-
-            Throwable err = walWriter.err;
-
-            if (err != null)
-                cctx.kernalContext().failure().process(new FailureContext(CRITICAL_ERROR, err));
-
-            if (expPos == UNCONDITIONAL_FLUSH)
-                expPos = (currentHandle().buf.tail());
-
-            Thread t = Thread.currentThread();
-
-            waiters.put(t, expPos);
-
-            LockSupport.unpark(walWriter.runner());
-
-            while (true) {
-                Long val = waiters.get(t);
-
-                assert val != null : "Only this thread can remove thread from waiters";
-
-                if (val == Long.MIN_VALUE) {
-                    waiters.remove(t);
-
-                    Throwable walWriterError = walWriter.err;
-
-                    if (walWriterError != null)
-                        throw new IgniteCheckedException("Flush buffer failed.", walWriterError);
-
-                    return;
-                }
-                else
-                    LockSupport.park();
-            }
-        }
-
-        /**
-         * @param pos Position in file to start write from. May be checked against actual position to wait previous
-         * writes to complete
-         * @param buf Buffer to write to file
-         * @throws StorageException If failed.
-         * @throws IgniteCheckedException If failed.
-         */
-        @SuppressWarnings("TooBroadScope")
-        private void writeBuffer(long pos, ByteBuffer buf) throws StorageException, IgniteCheckedException {
-            FileWriteHandle hdl = currentHandle();
-
-            assert hdl.fileIO != null : "Writing to a closed segment.";
-
-            checkNode();
-
-            long lastLogged = U.currentTimeMillis();
-
-            long logBackoff = 2_000;
-
-            // If we were too fast, need to wait previous writes to complete.
-            while (hdl.written != pos) {
-                assert hdl.written < pos : "written = " + hdl.written + ", pos = " + pos; // No one can write further than we are now.
-
-                // Permutation occurred between blocks write operations.
-                // Order of acquiring lock is not the same as order of write.
-                long now = U.currentTimeMillis();
-
-                if (now - lastLogged >= logBackoff) {
-                    if (logBackoff < 60 * 60_000)
-                        logBackoff *= 2;
-
-                    U.warn(log, "Still waiting for a concurrent write to complete [written=" + hdl.written +
-                        ", pos=" + pos + ", lastFsyncPos=" + hdl.lastFsyncPos + ", stop=" + hdl.stop.get() +
-                        ", actualPos=" + hdl.safePosition() + ']');
-
-                    lastLogged = now;
-                }
-
-                checkNode();
-            }
-
-            // Do the write.
-            int size = buf.remaining();
-
-            assert size > 0 : size;
-
-            try {
-                assert hdl.written == hdl.fileIO.position();
-
-                hdl.written += hdl.fileIO.writeFully(buf);
-
-                metrics.onWalBytesWritten(size);
-
-                assert hdl.written == hdl.fileIO.position();
-            }
-            catch (IOException e) {
-                StorageException se = new StorageException("Failed to write buffer.", e);
-
-                cctx.kernalContext().failure().process(new FailureContext(CRITICAL_ERROR, se));
-
-                throw se;
-            }
-        }
-    }
-
-    /**
-     * Syncs WAL segment file.
-     */
-    private class WalSegmentSyncer extends GridWorker {
-        /** Sync timeout. */
-        long syncTimeout;
-
-        /**
-         * @param igniteInstanceName Ignite instance name.
-         * @param log Logger.
-         */
-        public WalSegmentSyncer(String igniteInstanceName, IgniteLogger log) {
-            super(igniteInstanceName, "wal-segment-syncer", log);
-
-            syncTimeout = Math.max(IgniteSystemProperties.getLong(IGNITE_WAL_SEGMENT_SYNC_TIMEOUT,
-                DFLT_WAL_SEGMENT_SYNC_TIMEOUT), 100L);
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException {
-            while (!isCancelled()) {
-                sleep(syncTimeout);
-
-                try {
-                    flush(null, true);
-                }
-                catch (IgniteCheckedException e) {
-                    U.error(log, "Exception when flushing WAL.", e);
-                }
-            }
-        }
-
-        /** Shutted down the worker. */
-        private void shutdown() {
-            synchronized (this) {
-                U.cancel(this);
-            }
-
-            U.join(this, log);
-        }
-    }
-
-    /**
      * Scans provided folder for a WAL segment files
      * @param walFilesDir directory to scan
      * @return found WAL file descriptors


Mime
View raw message