You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2022/09/08 12:12:34 UTC
[GitHub] [ignite-3] SammyVimes commented on a diff in pull request #1061: IGNITE-17611 Local storage recovery code for TxStateRocksDbStorage
SammyVimes commented on code in PR #1061:
URL: https://github.com/apache/ignite-3/pull/1061#discussion_r965772345
##########
modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/flush/RocksDbFlusher.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.rocksdb.flush;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.rocksdb.AbstractEventListener;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Helper class do deal with RocksDB flushes. Provides an ability to wait until current state of data is flushed to the storage.
Review Comment:
```suggestion
* Helper class to deal with RocksDB flushes. Provides an ability to wait until current state of data is flushed to the storage.
```
##########
modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/flush/RocksDbFlusher.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.rocksdb.flush;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.rocksdb.AbstractEventListener;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Helper class do deal with RocksDB flushes. Provides an ability to wait until current state of data is flushed to the storage.
+ * Requires enabled {@link Options#setAtomicFlush(boolean)} option to work properly.
+ */
+public class RocksDbFlusher {
+ /** Logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(RocksDbFlusher.class);
+
+ /** Rocks DB instance. */
+ private volatile RocksDB db;
+
+ /** List of all column families. */
+ private volatile List<ColumnFamilyHandle> columnFamilyHandles;
+
+ /** Scheduled pool the schedule flushes. */
+ private final ScheduledExecutorService scheduledPool;
+
+ /** Thread pool to complete flush completion futures. */
+ final ExecutorService threadPool;
+
+ /** Supplier of delay values to batch independant flush requests. */
+ private final IntSupplier delaySupplier;
+
+ /** Flush completion callback. */
+ final Runnable onFlushCompleted;
+
+ /**
+ * Flush options to be used to asynchronously flush the Rocks DB memtable. It needs to be cached, because
+ * {@link RocksDB#flush(FlushOptions)} contract requires this object to not be GC-ed.
+ */
+ private final FlushOptions flushOptions = new FlushOptions().setWaitForFlush(false);
+
+ /** Map with flush futures by sequence number at the time of the {@link #awaitFlush(boolean)} call. */
+ final ConcurrentMap<Long, CompletableFuture<Void>> flushFuturesBySequenceNumber = new ConcurrentHashMap<>();
+
+ /** Latest known sequence number for persisted data. Not volatile, protected by explicit synchronization. */
+ private long latestPersistedSequenceNumber;
+
+ /** Mutex for {@link #latestPersistedSequenceNumber} modifications. */
+ private final Object latestPersistedSequenceNumberMux = new Object();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock;
+
+ /**
+ * Instance of the latest scheduled flush closure.
+ *
+ * @see #scheduleFlush()
+ */
+ private volatile Runnable latestFlushClosure;
+
+ /**
+ * Constructor.
+ *
+ * @param busyLock Busy lock.
+ * @param scheduledPool Scheduled pool the schedule flushes.
+ * @param threadPool Thread pool to run flush completion closure, provided by {@code onFlushCompleted} parameter.
+ * @param delaySupplier Supplier of delay values to batch independant flush requests. When {@link #awaitFlush(boolean)} is called with
Review Comment:
```suggestion
* @param delaySupplier Supplier of delay values to batch independent flush requests. When {@link #awaitFlush(boolean)} is called with
```
##########
modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java:
##########
@@ -193,6 +169,14 @@ public TableConfiguration configuration() {
/** {@inheritDoc} */
@Override
public void start() throws StorageException {
+ flusher = new RocksDbFlusher(
+ busyLock,
+ engine.scheduledPool(),
+ engine().threadPool(),
+ engine.configuration().flushDelayMillis()::value,
Review Comment:
This is ok, but I wonder, why not pass value itself? So that we can change configuration in runtime?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionAccess.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.table.distributed.raft.snapshot;
+
+/**
+ * Small abstractions for partition storages that includes only methods, mandatory for the snapshot storage.
+ */
+public interface PartitionAccess {
+ /**
+ * Returns persisted RAFT index for the partition.
Review Comment:
I liked it better when you called it sequence number. More agnostic
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbStorage.java:
##########
@@ -215,20 +392,21 @@ public TxStateRocksDbStorage(Path dbPath, ExecutorService snapshotExecutor) {
}
}
- /** {@inheritDoc} */
- @Override public CompletableFuture<Void> snapshot(Path snapshotPath) {
- return snapshotManager.createSnapshot(snapshotPath);
+ private static void throwStorageStoppedException() {
+ throw new IgniteInternalException(TX_STATE_STORAGE_STOPPED_ERR, "Transaction state storage is stopped");
}
- /** {@inheritDoc} */
- @Override public void restoreSnapshot(Path snapshotPath) {
- synchronized (snapshotRestoreLock) {
- destroy();
+ private byte[] uuidToBytes(UUID uuid) {
+ return ByteBuffer.allocate(2 * Long.BYTES).order(ByteOrder.BIG_ENDIAN)
+ .putLong(uuid.getMostSignificantBits())
+ .putLong(uuid.getLeastSignificantBits())
+ .array();
+ }
- start();
+ private UUID bytesToUuid(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN);
Review Comment:
Lazy!
##########
modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/BusyRocksIteratorAdapter.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.rocksdb;
+
+import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.rocksdb.RocksIterator;
+
+/**
+ * Adapter from a {@link RocksIterator} to a {@link Cursor} that also handles stopping of the storage.
+ */
+public abstract class BusyRocksIteratorAdapter<T> extends RocksIteratorAdapter<T> {
+ /** Busy lock. */
+ private final IgniteSpinBusyLock busyLock;
+
+ /**
+ * Constructor.
+ *
+ * @param busyLock Busy lock.
+ * @param it RocksDB iterator.
+ */
+ protected BusyRocksIteratorAdapter(IgniteSpinBusyLock busyLock, RocksIterator it) {
+ super(it);
+ this.busyLock = busyLock;
+ }
+
+ /**
+ * Handles busy lock acquiring failure. This means that db has been stopped and cursor can't proceed. Must throw an exception.
+ */
+ protected abstract void handleBusy();
+
+ @Override
+ public boolean hasNext() {
+ if (!busyLock.enterBusy()) {
+ handleBusy();
Review Comment:
If it must throw an exception, we should add something like `assert false` here. Or better yet make a private method (like `handleBusy0`) and call `handleBusy` from there (with assert false after the call) reducing copy-pasted code
##########
modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/flush/RocksDbFlusher.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.rocksdb.flush;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.rocksdb.AbstractEventListener;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Helper class do deal with RocksDB flushes. Provides an ability to wait until current state of data is flushed to the storage.
+ * Requires enabled {@link Options#setAtomicFlush(boolean)} option to work properly.
+ */
+public class RocksDbFlusher {
+ /** Logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(RocksDbFlusher.class);
+
+ /** Rocks DB instance. */
+ private volatile RocksDB db;
+
+ /** List of all column families. */
+ private volatile List<ColumnFamilyHandle> columnFamilyHandles;
+
+ /** Scheduled pool the schedule flushes. */
+ private final ScheduledExecutorService scheduledPool;
+
+ /** Thread pool to complete flush completion futures. */
+ final ExecutorService threadPool;
+
+ /** Supplier of delay values to batch independant flush requests. */
+ private final IntSupplier delaySupplier;
+
+ /** Flush completion callback. */
+ final Runnable onFlushCompleted;
+
+ /**
+ * Flush options to be used to asynchronously flush the Rocks DB memtable. It needs to be cached, because
+ * {@link RocksDB#flush(FlushOptions)} contract requires this object to not be GC-ed.
+ */
+ private final FlushOptions flushOptions = new FlushOptions().setWaitForFlush(false);
+
+ /** Map with flush futures by sequence number at the time of the {@link #awaitFlush(boolean)} call. */
+ final ConcurrentMap<Long, CompletableFuture<Void>> flushFuturesBySequenceNumber = new ConcurrentHashMap<>();
+
+ /** Latest known sequence number for persisted data. Not volatile, protected by explicit synchronization. */
+ private long latestPersistedSequenceNumber;
+
+ /** Mutex for {@link #latestPersistedSequenceNumber} modifications. */
+ private final Object latestPersistedSequenceNumberMux = new Object();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock;
+
+ /**
+ * Instance of the latest scheduled flush closure.
+ *
+ * @see #scheduleFlush()
+ */
+ private volatile Runnable latestFlushClosure;
+
+ /**
+ * Constructor.
+ *
+ * @param busyLock Busy lock.
+ * @param scheduledPool Scheduled pool the schedule flushes.
+ * @param threadPool Thread pool to run flush completion closure, provided by {@code onFlushCompleted} parameter.
+ * @param delaySupplier Supplier of delay values to batch independant flush requests. When {@link #awaitFlush(boolean)} is called with
+ * {@code true} parameter, the flusher waits given number of milliseconds (using {@code scheduledPool}) and then executes flush
+ * only if there were no other {@code awaitFlush(true)} calls. Otherwise, it does nothing after the timeout. This guarantees that
+ * either the last one wins, or automatic flush wins if there's an enlless stream of {@code awaitFlush(true)} calls with very small
+ * time-intervals between them. Such behavior allows to save on unnecessary flushes when multiple await flush calls appear at
+ * roughly the same time from different threads. For example, several partitions might be flushed at the same time, because they
+ * started at the same time and their flush frequency is also the same.
+ * @param onFlushCompleted Flush completion callback. Executed on every individual column family flush.
+ */
+ public RocksDbFlusher(
+ IgniteSpinBusyLock busyLock,
+ ScheduledExecutorService scheduledPool,
+ ExecutorService threadPool,
+ IntSupplier delaySupplier,
+ Runnable onFlushCompleted
+ ) {
+ this.busyLock = busyLock;
+ this.scheduledPool = scheduledPool;
+ this.threadPool = threadPool;
+ this.delaySupplier = delaySupplier;
+ this.onFlushCompleted = onFlushCompleted;
+ }
+
+ /**
+ * Returns an instance of {@link AbstractEventListener} to process actual RocksDB events. Returned listener must be set into
+ * {@link Options#setListeners(List)} before database is started. Otherwise, no events would occurre.
+ */
+ public AbstractEventListener listener() {
+ return new RocksDbFlushListener(this);
+ }
+
+ /**
+ * Initializes the flusher with DB instance and a list of column families.
+ *
+ * @param db Rocks DB instance.
+ * @param columnFamilyHandles List of all column families. Column families missing from this list may not have flush events processed.
+ */
+ @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
+ public void init(RocksDB db, List<ColumnFamilyHandle> columnFamilyHandles) {
+ this.db = db;
+ this.columnFamilyHandles = columnFamilyHandles;
+
+ synchronized (latestPersistedSequenceNumberMux) {
+ latestPersistedSequenceNumber = db.getLatestSequenceNumber();
+ }
+ }
+
+ /**
+ * Returns a future to wait next flush operation from the current point in time. Uses {@link RocksDB#getLatestSequenceNumber()} to
+ * achieve this, by fixing its value at the time of invokation. Storage is considered flushed when at least one persisted column
+ * family has its latest sequence number greater or equal to the one that we fixed. This is enough to guarantee that all column families
+ * have up-to-data state as well, because fluher expects its users to also have {@link Options#setAtomicFlush(boolean)} option
+ * enabled.
+ *
+ * @param schedule {@code true} if {@link RocksDB#flush(FlushOptions)} should be explicitly triggerred in the near future. Please refer
+ * to {@link RocksDbFlusher#RocksDbFlusher(IgniteSpinBusyLock, ScheduledExecutorService, ExecutorService, IntSupplier, Runnable)}
+ * parameters description to see what's really happening in this case.
+ *
+ * @see #scheduleFlush()
+ */
+ public CompletableFuture<Void> awaitFlush(boolean schedule) {
+ CompletableFuture<Void> future;
+
+ long dbSequenceNumber = db.getLatestSequenceNumber();
+
+ synchronized (latestPersistedSequenceNumberMux) {
+ if (dbSequenceNumber <= latestPersistedSequenceNumber) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ future = flushFuturesBySequenceNumber.computeIfAbsent(dbSequenceNumber, s -> new CompletableFuture<>());
+ }
+
+ if (schedule) {
+ scheduleFlush();
+ }
+
+ return future;
+ }
+
+ /**
+ * Schedules a flush of the table. If run several times within a small amount of time, only the last scheduled flush will be executed.
+ */
+ private void scheduleFlush() {
+ Runnable newClosure = new Runnable() {
+ @Override
+ public void run() {
+ if (latestFlushClosure != this) {
+ return;
+ }
+
+ if (!busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ // Explicit list of CF handles is mandatory!
+ // Default flush is buggy and only invokes listener methods for a single random CF.
+ db.flush(flushOptions, columnFamilyHandles);
+ } catch (RocksDBException e) {
+ LOG.error("Error occurred during the explicit flush", e);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+ };
+
+ latestFlushClosure = newClosure;
+
+ scheduledPool.schedule(newClosure, delaySupplier.getAsInt(), TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Completes all futures in {@link #flushFuturesBySequenceNumber} up to a given sequence number.
+ */
+ void completeFutures(long sequenceNumber) {
+ synchronized (latestPersistedSequenceNumberMux) {
+ if (sequenceNumber <= latestPersistedSequenceNumber) {
+ return;
+ }
+
+ latestPersistedSequenceNumber = sequenceNumber;
+ }
+
+ Set<Entry<Long, CompletableFuture<Void>>> entries = flushFuturesBySequenceNumber.entrySet();
+
+ for (Iterator<Entry<Long, CompletableFuture<Void>>> iterator = entries.iterator(); iterator.hasNext(); ) {
Review Comment:
this looks like a job for `ConcurrentSkipListMap`
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -751,7 +751,8 @@ private RaftGroupOptions groupOptionsForPartition(
//TODO Revisit peers String representation: https://issues.apache.org/jira/browse/IGNITE-17420
raftGroupOptions.snapshotStorageFactory(new PartitionSnapshotStorageFactory(
- partitionStorage,
+ //TODO IGNITE-17302 Use miniumum from mv storage and tx state storage.
Review Comment:
I thought usage of the minimum should be in this ticket (according to the ticket description)
##########
modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/flush/RocksDbFlusher.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.rocksdb.flush;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.rocksdb.AbstractEventListener;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Helper class do deal with RocksDB flushes. Provides an ability to wait until current state of data is flushed to the storage.
+ * Requires enabled {@link Options#setAtomicFlush(boolean)} option to work properly.
+ */
+public class RocksDbFlusher {
+ /** Logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(RocksDbFlusher.class);
+
+ /** Rocks DB instance. */
+ private volatile RocksDB db;
+
+ /** List of all column families. */
+ private volatile List<ColumnFamilyHandle> columnFamilyHandles;
+
+ /** Scheduled pool the schedule flushes. */
+ private final ScheduledExecutorService scheduledPool;
+
+ /** Thread pool to complete flush completion futures. */
+ final ExecutorService threadPool;
+
+ /** Supplier of delay values to batch independant flush requests. */
Review Comment:
```suggestion
/** Supplier of delay values to batch independent flush requests. */
```
##########
modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java:
##########
@@ -351,18 +287,14 @@ public void stop() throws StorageException {
busyLock.block();
- for (CompletableFuture<Void> future : flushFuturesBySequenceNumber.values()) {
- future.cancel(false);
- }
-
List<AutoCloseable> resources = new ArrayList<>();
+ resources.add(flusher::stop);
Review Comment:
CF handles should be closed as well, I suppose
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbStorage.java:
##########
@@ -55,10 +69,23 @@
/**
* Tx state storage implementation based on RocksDB.
*/
-public class TxStateRocksDbStorage implements TxStateStorage, AutoCloseable {
+public class TxStateRocksDbStorage implements TxStateStorage {
+ static {
+ RocksDB.loadLibrary();
Review Comment:
I wonder why we add `loadLibrary` every time we use RocksDB. AFAIK, they have static `loadLibrary` in the RocksDB classes, so library should be loaded when we load any RocksDB class
##########
modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/flush/RocksDbFlusher.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.rocksdb.flush;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.rocksdb.AbstractEventListener;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Helper class do deal with RocksDB flushes. Provides an ability to wait until current state of data is flushed to the storage.
+ * Requires enabled {@link Options#setAtomicFlush(boolean)} option to work properly.
+ */
+public class RocksDbFlusher {
+ /** Logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(RocksDbFlusher.class);
+
+ /** Rocks DB instance. */
+ private volatile RocksDB db;
+
+ /** List of all column families. */
+ private volatile List<ColumnFamilyHandle> columnFamilyHandles;
+
+ /** Scheduled pool the schedule flushes. */
Review Comment:
```suggestion
/** Scheduled pool to schedule flushes. */
```
##########
modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/flush/RocksDbFlusher.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.rocksdb.flush;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.rocksdb.AbstractEventListener;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Helper class do deal with RocksDB flushes. Provides an ability to wait until current state of data is flushed to the storage.
+ * Requires enabled {@link Options#setAtomicFlush(boolean)} option to work properly.
+ */
+public class RocksDbFlusher {
+ /** Logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(RocksDbFlusher.class);
+
+ /** Rocks DB instance. */
+ private volatile RocksDB db;
+
+ /** List of all column families. */
+ private volatile List<ColumnFamilyHandle> columnFamilyHandles;
+
+ /** Scheduled pool the schedule flushes. */
+ private final ScheduledExecutorService scheduledPool;
+
+ /** Thread pool to complete flush completion futures. */
+ final ExecutorService threadPool;
+
+ /** Supplier of delay values to batch independant flush requests. */
+ private final IntSupplier delaySupplier;
+
+ /** Flush completion callback. */
+ final Runnable onFlushCompleted;
+
+ /**
+ * Flush options to be used to asynchronously flush the Rocks DB memtable. It needs to be cached, because
+ * {@link RocksDB#flush(FlushOptions)} contract requires this object to not be GC-ed.
+ */
+ private final FlushOptions flushOptions = new FlushOptions().setWaitForFlush(false);
+
+ /** Map with flush futures by sequence number at the time of the {@link #awaitFlush(boolean)} call. */
+ final ConcurrentMap<Long, CompletableFuture<Void>> flushFuturesBySequenceNumber = new ConcurrentHashMap<>();
+
+ /** Latest known sequence number for persisted data. Not volatile, protected by explicit synchronization. */
+ private long latestPersistedSequenceNumber;
+
+ /** Mutex for {@link #latestPersistedSequenceNumber} modifications. */
+ private final Object latestPersistedSequenceNumberMux = new Object();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock;
+
+ /**
+ * Instance of the latest scheduled flush closure.
+ *
+ * @see #scheduleFlush()
+ */
+ private volatile Runnable latestFlushClosure;
+
+ /**
+ * Constructor.
+ *
+ * @param busyLock Busy lock.
+ * @param scheduledPool Scheduled pool the schedule flushes.
+ * @param threadPool Thread pool to run flush completion closure, provided by {@code onFlushCompleted} parameter.
+ * @param delaySupplier Supplier of delay values to batch independant flush requests. When {@link #awaitFlush(boolean)} is called with
+ * {@code true} parameter, the flusher waits given number of milliseconds (using {@code scheduledPool}) and then executes flush
+ * only if there were no other {@code awaitFlush(true)} calls. Otherwise, it does nothing after the timeout. This guarantees that
+ * either the last one wins, or automatic flush wins if there's an enlless stream of {@code awaitFlush(true)} calls with very small
+ * time-intervals between them. Such behavior allows to save on unnecessary flushes when multiple await flush calls appear at
+ * roughly the same time from different threads. For example, several partitions might be flushed at the same time, because they
+ * started at the same time and their flush frequency is also the same.
+ * @param onFlushCompleted Flush completion callback. Executed on every individual column family flush.
+ */
+ public RocksDbFlusher(
+ IgniteSpinBusyLock busyLock,
+ ScheduledExecutorService scheduledPool,
+ ExecutorService threadPool,
+ IntSupplier delaySupplier,
+ Runnable onFlushCompleted
+ ) {
+ this.busyLock = busyLock;
+ this.scheduledPool = scheduledPool;
+ this.threadPool = threadPool;
+ this.delaySupplier = delaySupplier;
+ this.onFlushCompleted = onFlushCompleted;
+ }
+
+ /**
+ * Returns an instance of {@link AbstractEventListener} to process actual RocksDB events. Returned listener must be set into
+ * {@link Options#setListeners(List)} before database is started. Otherwise, no events would occurre.
+ */
+ public AbstractEventListener listener() {
+ return new RocksDbFlushListener(this);
+ }
+
+ /**
+ * Initializes the flusher with DB instance and a list of column families.
+ *
+ * @param db Rocks DB instance.
+ * @param columnFamilyHandles List of all column families. Column families missing from this list may not have flush events processed.
+ */
+ @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
+ public void init(RocksDB db, List<ColumnFamilyHandle> columnFamilyHandles) {
+ this.db = db;
+ this.columnFamilyHandles = columnFamilyHandles;
+
+ synchronized (latestPersistedSequenceNumberMux) {
+ latestPersistedSequenceNumber = db.getLatestSequenceNumber();
+ }
+ }
+
+ /**
+ * Returns a future to wait next flush operation from the current point in time. Uses {@link RocksDB#getLatestSequenceNumber()} to
+ * achieve this, by fixing its value at the time of invokation. Storage is considered flushed when at least one persisted column
+ * family has its latest sequence number greater or equal to the one that we fixed. This is enough to guarantee that all column families
+ * have up-to-data state as well, because fluher expects its users to also have {@link Options#setAtomicFlush(boolean)} option
Review Comment:
```suggestion
* have up-to-data state as well, because flusher expects its users to also have {@link Options#setAtomicFlush(boolean)} option
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org