| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.ignite.internal.processors.cache.persistence.snapshot.dump; |
| |
| import java.io.Closeable; |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.Serializable; |
| import java.nio.ByteBuffer; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.LongAdder; |
| import java.util.concurrent.locks.LockSupport; |
| import java.util.stream.Collectors; |
| import org.apache.ignite.IgniteCache; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteDataStreamer; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.dump.DumpEntry; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.processors.cache.CacheGroupContext; |
| import org.apache.ignite.internal.processors.cache.CacheObject; |
| import org.apache.ignite.internal.processors.cache.CacheObjectContext; |
| import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; |
| import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManager; |
| import org.apache.ignite.internal.processors.cache.KeyCacheObject; |
| import org.apache.ignite.internal.processors.cache.StoredCacheData; |
| import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; |
| import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow; |
| import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; |
| import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; |
| import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId; |
| import org.apache.ignite.internal.processors.cache.persistence.snapshot.AbstractCreateSnapshotFutureTask; |
| import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager; |
| import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFutureTaskResult; |
| import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotSender; |
| import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusIO; |
| import org.apache.ignite.internal.processors.cache.tree.CacheDataTree; |
| import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; |
| import org.apache.ignite.internal.processors.cache.version.GridCacheVersionManager; |
| import org.apache.ignite.internal.util.BasicRateLimiter; |
| import org.apache.ignite.internal.util.GridConcurrentHashSet; |
| import org.apache.ignite.internal.util.IgniteUtils; |
| import org.apache.ignite.internal.util.lang.GridCloseableIterator; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION; |
| import static org.apache.ignite.internal.processors.cache.GridLocalConfigManager.cacheDataFilename; |
| import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DIR_PREFIX; |
| import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_GRP_DIR_PREFIX; |
| import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.DUMP_LOCK; |
| import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.Dump.dumpPartFileName; |
| import static org.apache.ignite.internal.util.IgniteUtils.toLong; |
| |
| /** |
| * Task creates cache group dump. |
| * Dump is a consistent snapshot of cache entries. |
| * Directories structure is same as a full snapshot but each partitions saved in "part-0.dump" file. |
| * Files structure is a set of {@link DumpEntry} written one by one. |
| * |
| * @see Dump |
| * @see DumpEntry |
| */ |
| public class CreateDumpFutureTask extends AbstractCreateSnapshotFutureTask implements DumpEntryChangeListener { |
| /** Dump files name. */ |
| public static final String DUMP_FILE_EXT = ".dump"; |
| |
| /** Root dump directory. */ |
| private final File dumpDir; |
| |
| /** */ |
| private final FileIOFactory ioFactory; |
| |
| /** If {@code true} then compress partition files. */ |
| private final boolean compress; |
| |
| /** Dump transfer rate limiter. */ |
| private final BasicRateLimiter rateLimiter; |
| |
| /** Processed dump size in bytes. */ |
| private final AtomicLong processedSize = new AtomicLong(); |
| |
| /** If {@code null} then encryption disabled. */ |
| private final @Nullable Serializable encKey; |
| |
| /** |
| * Dump contextes. |
| * Key is [group_id, partition_id] combined in single long value. |
| * |
| * @see IgniteUtils#toLong(int, int) |
| */ |
| private final Map<Long, PartitionDumpContext> dumpCtxs = new ConcurrentHashMap<>(); |
| |
| /** Local node is primary for set of group partitions. */ |
| private final Map<Integer, Set<Integer>> grpPrimaries = new ConcurrentHashMap<>(); |
| |
| /** |
| * Map shared across all instances of {@link PartitionDumpContext} and {@link DumpEntrySerializer}. |
| * We use per thread buffer because number of threads is fewer then number of partitions. |
| * Regular count of partitions is {@link RendezvousAffinityFunction#DFLT_PARTITION_COUNT} |
| * and thread is {@link IgniteConfiguration#DFLT_PUBLIC_THREAD_CNT} whic is significantly less. |
| */ |
| private final ConcurrentMap<Long, ByteBuffer> thLocBufs = new ConcurrentHashMap<>(); |
| |
| /** |
| * Map shared across all instances of {@link PartitionDumpContext} and {@link DumpEntrySerializer}. |
| * Used to store encrypted data in case dump encrypted. |
| */ |
| private final @Nullable ConcurrentMap<Long, ByteBuffer> encThLocBufs; |
| |
| /** |
| * Cache entry version at the dump start. |
| * Regular updates with {@link IgniteCache#put(Object, Object)} and similar calls use version generated with |
| * {@link GridCacheVersionManager#next(GridCacheVersion)}. This version value monotonically grows. It is generated |
| * on a <b>primary</b> node and is <b>propagated</b> to the backups. So, on a primary we can distinguish updates |
| * that happen before and after the dump start. |
| */ |
| private GridCacheVersion startVer; |
| |
| /** |
| * @param cctx Cache context. |
| * @param srcNodeId Node id which cause snapshot task creation. |
| * @param reqId Snapshot operation request ID. |
| * @param dumpName Dump name. |
| * @param ioFactory IO factory. |
| * @param snpSndr Snapshot sender. |
| * @param rateLimiter Dump transfer rate limiter. |
| * @param parts Parts to dump. |
| * @param compress If {@code true} then compress partition files. |
| * @param encrypt If {@code true} then content of dump encrypted. |
| */ |
| public CreateDumpFutureTask( |
| GridCacheSharedContext<?, ?> cctx, |
| UUID srcNodeId, |
| UUID reqId, |
| String dumpName, |
| File dumpDir, |
| FileIOFactory ioFactory, |
| BasicRateLimiter rateLimiter, |
| SnapshotSender snpSndr, |
| Map<Integer, Set<Integer>> parts, |
| boolean compress, |
| boolean encrypt |
| ) { |
| super( |
| cctx, |
| srcNodeId, |
| reqId, |
| dumpName, |
| snpSndr, |
| parts |
| ); |
| |
| this.dumpDir = dumpDir; |
| |
| this.ioFactory = compress ? new WriteOnlyZipFileIOFactory(ioFactory) : new BufferedFileIOFactory(ioFactory); |
| |
| this.compress = compress; |
| this.rateLimiter = rateLimiter; |
| this.encKey = encrypt ? cctx.gridConfig().getEncryptionSpi().create() : null; |
| this.encThLocBufs = encrypt ? new ConcurrentHashMap<>() : null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean start() { |
| try { |
| if (log.isInfoEnabled()) |
| log.info("Start cache dump [name=" + snpName + ", grps=" + parts.keySet() + ']'); |
| |
| createDumpLock(); |
| |
| processPartitions(); |
| |
| prepare(); |
| |
| saveSnapshotData(); |
| } |
| catch (IgniteCheckedException | IOException e) { |
| acceptException(e); |
| } |
| |
| return false; // Don't wait for checkpoint. |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void processPartitions() throws IgniteCheckedException { |
| super.processPartitions(); |
| |
| processed.values().forEach(parts -> parts.remove(INDEX_PARTITION)); |
| } |
| |
| /** Prepares all data structures to dump entries. */ |
| private void prepare() throws IOException, IgniteCheckedException { |
| startVer = cctx.versions().next(cctx.kernalContext().discovery().topologyVersion()); |
| |
| for (Map.Entry<Integer, Set<Integer>> e : processed.entrySet()) { |
| int grp = e.getKey(); |
| |
| File grpDumpDir = groupDirectory(cctx.cache().cacheGroup(grp)); |
| |
| if (!grpDumpDir.mkdirs()) |
| throw new IgniteCheckedException("Dump directory can't be created: " + grpDumpDir); |
| |
| CacheGroupContext gctx = cctx.cache().cacheGroup(grp); |
| |
| for (GridCacheContext<?, ?> cctx : gctx.caches()) |
| cctx.dumpListener(this); |
| |
| grpPrimaries.put( |
| grp, |
| gctx.affinity().primaryPartitions(gctx.shared().kernalContext().localNodeId(), gctx.affinity().lastVersion()) |
| ); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected List<CompletableFuture<Void>> saveCacheConfigs() { |
| return processed.keySet().stream().map(grp -> runAsync(() -> { |
| CacheGroupContext gctx = cctx.cache().cacheGroup(grp); |
| |
| File grpDir = groupDirectory(gctx); |
| |
| IgniteUtils.ensureDirectory(grpDir, "dump group directory", null); |
| |
| for (GridCacheContext<?, ?> cacheCtx : gctx.caches()) { |
| DynamicCacheDescriptor desc = cctx.kernalContext().cache().cacheDescriptor(cacheCtx.cacheId()); |
| |
| StoredCacheData cacheData = new StoredCacheData(new CacheConfiguration(desc.cacheConfiguration())); |
| |
| cacheData.queryEntities(desc.schema().entities()); |
| cacheData.sql(desc.sql()); |
| |
| cctx.cache().configManager().writeCacheData(cacheData, new File(grpDir, cacheDataFilename(cacheData.config()))); |
| } |
| })).collect(Collectors.toList()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected List<CompletableFuture<Void>> saveGroup(int grp, Set<Integer> grpParts) { |
| long start = System.currentTimeMillis(); |
| |
| AtomicLong entriesCnt = new AtomicLong(); |
| AtomicLong writtenEntriesCnt = new AtomicLong(); |
| AtomicLong changedEntriesCnt = new AtomicLong(); |
| |
| String name = cctx.cache().cacheGroup(grp).cacheOrGroupName(); |
| |
| CacheGroupContext gctx = cctx.kernalContext().cache().cacheGroup(grp); |
| |
| if (log.isInfoEnabled()) |
| log.info("Start group dump [name=" + name + ", id=" + grp + ']'); |
| |
| List<CompletableFuture<Void>> futs = grpParts.stream().map(part -> runAsync(() -> { |
| long entriesCnt0 = 0; |
| long writtenEntriesCnt0 = 0; |
| |
| int coCtxCache = 0; |
| CacheObjectContext coCtx = null; |
| |
| try (PartitionDumpContext dumpCtx = dumpContext(grp, part)) { |
| try (GridCloseableIterator<CacheDataRow> rows = gctx.offheap().reservedIterator(part, dumpCtx.topVer)) { |
| if (rows == null) |
| throw new IgniteCheckedException("Partition missing [part=" + part + ']'); |
| |
| while (rows.hasNext()) { |
| CacheDataRow row = rows.next(); |
| |
| assert row.partition() == part; |
| |
| int cache = row.cacheId() == 0 ? grp : row.cacheId(); |
| |
| if (cache != coCtxCache) { |
| coCtxCache = cache; |
| |
| coCtx = cctx.cacheObjectContext(coCtxCache); |
| } |
| |
| if (dumpCtx.writeForIterator(cache, row.expireTime(), row.key(), row.value(), row.version(), coCtx)) |
| writtenEntriesCnt0++; |
| |
| entriesCnt0++; |
| } |
| } |
| catch (IOException e) { |
| throw new IgniteException(e); |
| } |
| |
| entriesCnt.addAndGet(entriesCnt0); |
| writtenEntriesCnt.addAndGet(writtenEntriesCnt0); |
| changedEntriesCnt.addAndGet(dumpCtx.changedCnt.intValue()); |
| |
| if (log.isDebugEnabled()) { |
| log.debug("Finish group partition dump [name=" + name + |
| ", id=" + grp + |
| ", part=" + part + |
| ", time=" + (System.currentTimeMillis() - start) + |
| ", iterEntriesCnt=" + entriesCnt + |
| ", writtenIterEntriesCnt=" + entriesCnt + |
| ", changedEntriesCnt=" + changedEntriesCnt + ']'); |
| } |
| } |
| })).collect(Collectors.toList()); |
| |
| int futsSize = futs.size(); |
| |
| CompletableFuture.allOf(futs.toArray(new CompletableFuture[futsSize])).whenComplete((res, t) -> { |
| clearDumpListener(gctx); |
| |
| if (log.isInfoEnabled()) { |
| log.info("Finish group dump [name=" + name + |
| ", id=" + grp + |
| ", time=" + (System.currentTimeMillis() - start) + |
| ", iterEntriesCnt=" + entriesCnt.get() + |
| ", writtenIterEntriesCnt=" + writtenEntriesCnt.get() + |
| ", changedEntriesCnt=" + changedEntriesCnt.get() + ']'); |
| } |
| }); |
| |
| return futs; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void beforeChange(GridCacheContext cctx, KeyCacheObject key, CacheObject val, long expireTime, GridCacheVersion ver) { |
| try { |
| int part = key.partition(); |
| int grp = cctx.groupId(); |
| |
| assert part != -1; |
| |
| if (!processed.get(grp).contains(part)) |
| return; |
| |
| dumpContext(grp, part).writeChanged(cctx.cacheId(), expireTime, key, val, ver); |
| } |
| catch (IgniteException | IgniteCheckedException | IOException e) { |
| acceptException(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected synchronized CompletableFuture<Void> closeAsync() { |
| if (closeFut == null) { |
| dumpCtxs.values().forEach(PartitionDumpContext::close); |
| |
| Throwable err0 = err.get(); |
| |
| Set<GroupPartitionId> taken = new HashSet<>(); |
| |
| for (Map.Entry<Integer, Set<Integer>> e : processed.entrySet()) { |
| int grp = e.getKey(); |
| |
| clearDumpListener(cctx.cache().cacheGroup(grp)); |
| |
| for (Integer part : e.getValue()) |
| taken.add(new GroupPartitionId(grp, part)); |
| } |
| |
| closeFut = CompletableFuture.runAsync( |
| () -> { |
| thLocBufs.clear(); |
| if (encThLocBufs != null) |
| encThLocBufs.clear(); |
| onDone(new SnapshotFutureTaskResult(taken, null), err0); |
| }, |
| cctx.kernalContext().pools().getSystemExecutorService() |
| ); |
| } |
| |
| return closeFut; |
| } |
| |
| /** @return Processed dump size in bytes. */ |
| public long processedSize() { |
| return processedSize.get(); |
| } |
| |
| /** */ |
| private void clearDumpListener(CacheGroupContext gctx) { |
| for (GridCacheContext<?, ?> cctx : gctx.caches()) |
| cctx.dumpListener(null); |
| } |
| |
| /** */ |
| private void createDumpLock() throws IgniteCheckedException, IOException { |
| File nodeDumpDir = IgniteSnapshotManager.nodeDumpDirectory(dumpDir, cctx); |
| |
| if (!nodeDumpDir.mkdirs()) |
| throw new IgniteCheckedException("Can't create node dump directory: " + nodeDumpDir.getAbsolutePath()); |
| |
| File lock = new File(nodeDumpDir, DUMP_LOCK); |
| |
| if (!lock.createNewFile()) |
| throw new IgniteCheckedException("Lock file can't be created or already exists: " + lock.getAbsolutePath()); |
| } |
| |
| /** */ |
| private PartitionDumpContext dumpContext(int grp, int part) { |
| return dumpCtxs.computeIfAbsent( |
| toLong(grp, part), |
| key -> new PartitionDumpContext(cctx.kernalContext().cache().cacheGroup(grp), part) |
| ); |
| } |
| |
| /** |
| * Context of dump single partition. |
| */ |
| private class PartitionDumpContext implements Closeable { |
| /** Group id. */ |
| private final int grp; |
| |
| /** Partition id. */ |
| private final int part; |
| |
| /** |
| * Key is cache id, values is set of keys dumped via |
| * {@link #writeChanged(int, long, KeyCacheObject, CacheObject, GridCacheVersion)}. |
| */ |
| private final Map<Integer, Set<KeyCacheObject>> changed; |
| |
| /** Cache id for {@link #iterLastKey} */ |
| private int iterLastKeyCache; |
| |
| /** Last key dumped via {@link #writeForIterator(int, long, KeyCacheObject, CacheObject, GridCacheVersion, CacheObjectContext)}. */ |
| private @Nullable KeyCacheObject iterLastKey; |
| |
| /** Count of entries changed during dump creation. */ |
| private final LongAdder changedCnt = new LongAdder(); |
| |
| /** Partition dump file. Lazily initialized to prevent creation files for empty partitions. */ |
| private final FileIO file; |
| |
| /** Topology Version. */ |
| private final AffinityTopologyVersion topVer; |
| |
| /** Partition serializer. */ |
| private final DumpEntrySerializer serializer; |
| |
| /** If {@code true} context is closed. */ |
| private volatile boolean closed; |
| |
| /** |
| * Count of writers. When count becomes {@code 0} context must be closed. |
| * By deafult, one writer exists - partition iterator. |
| * Each call of {@link #writeChanged(int, long, KeyCacheObject, CacheObject, GridCacheVersion)} increment writers count. |
| * When count of writers becomes zero we good to relase all resources associated with partition dump. |
| */ |
| private final AtomicInteger writers = new AtomicInteger(1); |
| |
| /** Primary node/partition flag. */ |
| private final boolean primary; |
| |
| /** |
| * @param gctx Group context. |
| * @param part Partition id. |
| */ |
| private PartitionDumpContext(CacheGroupContext gctx, int part) { |
| assert gctx != null; |
| |
| try { |
| this.part = part; |
| grp = gctx.groupId(); |
| topVer = gctx.topology().lastTopologyChangeVersion(); |
| |
| primary = grpPrimaries.get(gctx.groupId()).contains(part); |
| |
| serializer = new DumpEntrySerializer( |
| thLocBufs, |
| encThLocBufs, |
| encKey, |
| cctx.gridConfig().getEncryptionSpi() |
| ); |
| changed = new HashMap<>(); |
| |
| for (int cache : gctx.cacheIds()) |
| changed.put(cache, new GridConcurrentHashSet<>()); |
| |
| File dumpFile = new File(groupDirectory(gctx), dumpPartFileName(part, compress)); |
| |
| if (!dumpFile.createNewFile()) |
| throw new IgniteException("Dump file can't be created: " + dumpFile); |
| |
| file = ioFactory.create(dumpFile); |
| } |
| catch (IOException | IgniteCheckedException e) { |
| throw new IgniteException(e); |
| } |
| } |
| |
| /** |
| * Writes entry that changed while dump creation in progress. |
| * @param cache Cache id. |
| * @param expireTime Expire time. |
| * @param key Key. |
| * @param val Value before change. |
| * @param ver Version before change. |
| */ |
| public void writeChanged( |
| int cache, |
| long expireTime, |
| KeyCacheObject key, |
| CacheObject val, |
| GridCacheVersion ver |
| ) throws IgniteCheckedException, IOException { |
| String reasonToSkip = null; |
| |
| if (closed) // Quick exit. Partition already saved in dump. |
| reasonToSkip = "partition already saved"; |
| else { |
| writers.getAndIncrement(); |
| |
| try { |
| if (closed) // Partition already saved in dump. |
| reasonToSkip = "partition already saved"; |
| else if (afterStart(ver)) |
| reasonToSkip = "greater version"; |
| else { |
| CacheObjectContext coCtx = cctx.cacheObjectContext(cache); |
| |
| synchronized (serializer) { // Prevent concurrent access to the dump file. |
| if (writtenByIterator(cache, key, coCtx)) |
| reasonToSkip = "written by iterator"; // Saved by iterator, already. Skip. |
| else if (!changed.get(cache).add(key)) // Entry changed several time during dump. |
| reasonToSkip = "changed several times"; |
| else if (val == null) |
| // Previous value is null. Entry created after dump start, skip. |
| reasonToSkip = "newly created or already removed"; |
| else |
| write(cache, expireTime, key, val, ver, coCtx); |
| } |
| |
| if (reasonToSkip == null) |
| changedCnt.increment(); |
| } |
| } |
| finally { |
| writers.decrementAndGet(); |
| } |
| } |
| |
| if (log.isTraceEnabled()) { |
| log.trace("Listener [grp=" + grp + |
| ", cache=" + cache + |
| ", part=" + part + |
| ", key=" + key + |
| ", ver=" + ver + |
| ", written=" + (reasonToSkip == null ? "true" : reasonToSkip) + |
| ", startVer=" + (startVer != null) + ']'); |
| } |
| } |
| |
| /** |
| * Writes entry fetched from partition iterator. |
| * |
| * @param cache Cache id. |
| * @param expireTime Expire time. |
| * @param key Key. |
| * @param val Value. |
| * @param ver Version. |
| * @return {@code True} if entry was written in dump. |
| * {@code false} if it was already written by {@link #writeChanged(int, long, KeyCacheObject, CacheObject, GridCacheVersion)}. |
| */ |
| public boolean writeForIterator( |
| int cache, |
| long expireTime, |
| KeyCacheObject key, |
| CacheObject val, |
| GridCacheVersion ver, |
| CacheObjectContext coCtx |
| ) throws IgniteCheckedException, IOException { |
| String reason = null; |
| |
| if (afterStart(ver)) |
| reason = "greater version"; |
| else { |
| synchronized (serializer) { // Prevent concurrent access to the dump file. |
| iterLastKeyCache = cache; |
| iterLastKey = key; |
| |
| if (changed.get(cache).contains(key)) |
| reason = "written by listener"; |
| else |
| write(cache, expireTime, key, val, ver, coCtx); |
| } |
| } |
| |
| if (log.isTraceEnabled()) { |
| log.trace("Iterator [" + |
| "grp=" + grp + |
| ", cache=" + cache + |
| ", part=" + part + |
| ", key=" + key + |
| ", written=" + (reason == null ? "true" : reason) + |
| ", ver=" + ver + |
| ", startVer=" + (startVer != null) + ']'); |
| } |
| |
| return reason == null; |
| } |
| |
| /** */ |
| private void write( |
| int cache, |
| long expireTime, |
| KeyCacheObject key, |
| CacheObject val, |
| GridCacheVersion ver, |
| CacheObjectContext coCtx |
| ) throws IgniteCheckedException, IOException { |
| ByteBuffer buf = serializer.writeToBuffer(cache, expireTime, key, val, ver, coCtx); |
| |
| rateLimiter.acquire(buf.limit()); |
| |
| if (file.writeFully(buf) != buf.limit()) |
| throw new IgniteException("Can't write row"); |
| |
| processedSize.addAndGet(buf.limit()); |
| } |
| |
| /** |
| * Note, usage of {@link IgniteDataStreamer} may lead to dump inconsistency. |
| * Because, streamer use the same {@link GridCacheVersion} for all entries. |
| * So, we can't efficiently filter out new entries and write them all to dump. |
| * |
| * @param ver Entry version. |
| * @return {@code True} if {@code ver} appeared after dump started. |
| * @see GridCacheVersionManager#isolatedStreamerVersion() |
| */ |
| private boolean afterStart(GridCacheVersion ver) { |
| return primary && ver.isGreater(startVer); |
| } |
| |
| /** |
| * Iterator returned by {@link IgniteCacheOffheapManager#reservedIterator(int, AffinityTopologyVersion)} |
| * iterates key in ascending order set by {@link CacheDataTree#compare(BPlusIO, long, int, CacheSearchRow)}. |
| * So if key changed by the user (see {@link #writeChanged(int, long, KeyCacheObject, CacheObject, GridCacheVersion)}) |
| * is greater than last key written by partition iterator then it hasn't been saved in dump yet and must be written. |
| * Otherwise, key already saved by the iterator and must be skiped. |
| * |
| * @param cache Cache id. |
| * @param key Key to write with {@link #writeChanged(int, long, KeyCacheObject, CacheObject, GridCacheVersion)}. |
| * @return {@code True} if key written by iterator, already. {@code False} otherwise. |
| * @see CacheDataTree#compareBytes(byte[], byte[]) |
| */ |
| private boolean writtenByIterator(int cache, KeyCacheObject key, CacheObjectContext coCtx) throws IgniteCheckedException { |
| if (iterLastKey == null) |
| return false; |
| |
| int cmp = Integer.compare(iterLastKeyCache, cache); |
| |
| if (cmp != 0) |
| return cmp > 0; |
| |
| cmp = Integer.compare(iterLastKey.hashCode(), key.hashCode()); |
| |
| if (cmp != 0) |
| return cmp > 0; |
| |
| byte[] bytes1 = iterLastKey.valueBytes(coCtx); |
| byte[] bytes2 = key.valueBytes(coCtx); |
| |
| return CacheDataTree.compareBytes(bytes1, bytes2) >= 0; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void close() { |
| synchronized (this) { // Prevent concurrent close invocation. |
| if (closed) |
| return; |
| |
| closed = true; |
| } |
| |
| writers.decrementAndGet(); |
| |
| while (writers.get() > 0) // Waiting for all on the fly listeners to complete. |
| LockSupport.parkNanos(1_000_000); |
| |
| try { |
| file.close(); |
| } |
| catch (IOException e) { |
| throw new IgniteException(e); |
| } |
| } |
| } |
| |
| /** */ |
| private File groupDirectory(CacheGroupContext grpCtx) throws IgniteCheckedException { |
| return new File( |
| IgniteSnapshotManager.nodeDumpDirectory(dumpDir, cctx), |
| (grpCtx.caches().size() > 1 ? CACHE_GRP_DIR_PREFIX : CACHE_DIR_PREFIX) + grpCtx.cacheOrGroupName() |
| ); |
| } |
| |
| /** @return Encryption key. */ |
| public @Nullable Serializable encryptionKey() { |
| return encKey; |
| } |
| } |
