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 2021/05/12 12:02:37 UTC
[GitHub] [ignite-3] sk0x50 commented on a change in pull request #114: IGNITE-14666 Add proper listener handling in DistributedConfigurationStorage and LocalConfigurationStorage
sk0x50 commented on a change in pull request #114:
URL: https://github.com/apache/ignite-3/pull/114#discussion_r630896146
##########
File path: modules/configuration/src/main/java/org/apache/ignite/configuration/storage/Data.java
##########
@@ -56,14 +51,7 @@ public Data(Map<String, Serializable> values, long cfgVersion, long storageRevis
* Get version.
* @return version.
*/
- public long cfgVersion() {
- return cfgVersion;
- }
-
- /**
- * @return Storage revision.
- */
- public long storageRevision() {
- return storageRevision;
+ public long changeId() {
Review comment:
It seems to me, that the provided Javadoc looks too "concise". Moreover, the `changeId` name does not match `version`. It would be nice to add a description of this "version". Is it a version of distributed meta storage? Does this "version" correspond to the "updated" version of a configuration?
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/DistributedConfigurationStorage.java
##########
@@ -49,56 +84,178 @@ public DistributedConfigurationStorage(MetaStorageManager metaStorageMgr) {
this.metaStorageMgr = metaStorageMgr;
}
- /** Map to store values. */
- private Map<String, Serializable> map = new ConcurrentHashMap<>();
-
/** Change listeners. */
private List<ConfigurationStorageListener> listeners = new CopyOnWriteArrayList<>();
- /** Storage version. */
- private AtomicLong version = new AtomicLong(0);
+ /** Storage version. It stores actual metastorage revision, that is applied to configuration manager. */
+ private AtomicLong ver = new AtomicLong(0L);
/** {@inheritDoc} */
@Override public synchronized Data readAll() throws StorageException {
- return new Data(new HashMap<>(map), version.get(), 0);
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ Iterator<Entry> entries = allDstCfgKeys().iterator();
+
+ long maxRevision = 0L;
+
+ if (!entries.hasNext())
+ return new Data(data, ver.get());
+
+ Entry entryForMasterKey = entries.next();
+
+ // First key must be the masterKey because it's supposed to be the first in lexicographical order
+ assert entryForMasterKey.key().equals(MASTER_KEY);
+
+ while (entries.hasNext()) {
+ Entry entry = entries.next();
+
+ data.put(entry.key().toString().substring((DISTRIBUTED_PREFIX).length()), (Serializable)ByteUtils.fromBytes(entry.value()));
+
+ // Move to stream
+ if (maxRevision < entry.revision())
+ maxRevision = entry.revision();
+
+ }
+
+ if (!data.isEmpty()) {
+ assert maxRevision == entryForMasterKey.revision();
+
+ assert maxRevision >= ver.get();
+
+ return new Data(data, maxRevision);
+ }
+
+ return new Data(data, ver.get());
}
/** {@inheritDoc} */
- @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long version) {
- if (version != this.version.get())
+ @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long sentVersion) {
+ assert sentVersion <= ver.get();
+
+ if (sentVersion != ver.get())
+ // This means that sentVersion is less than version and other node has already updated configuration and
+ // write should be retried. Actual version will be set when watch and corresponding configuration listener
+ // updates configuration and notifyApplied is triggered afterwards.
return CompletableFuture.completedFuture(false);
+ HashSet<Operation> operations = new HashSet<>();
+
for (Map.Entry<String, Serializable> entry : newValues.entrySet()) {
+ Key key = new Key(DISTRIBUTED_PREFIX + entry.getKey());
+
if (entry.getValue() != null)
- map.put(entry.getKey(), entry.getValue());
+ // TODO: investigate overhead when serialize int, long, double, boolean, string, arrays of above
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-14698
+ operations.add(Operations.put(key, ByteUtils.toBytes(entry.getValue())));
else
- map.remove(entry.getKey());
+ operations.add(Operations.remove(key));
}
- this.version.incrementAndGet();
-
- listeners.forEach(listener -> listener.onEntriesChanged(new Data(newValues, this.version.get(), 0)));
+ operations.add(Operations.put(MASTER_KEY, ByteUtils.longToBytes(sentVersion)));
- return CompletableFuture.completedFuture(true);
+ return metaStorageMgr.invoke(
+ Conditions.key(MASTER_KEY).revision().eq(ver.get()),
+ operations,
+ Collections.singleton(Operations.noop()));
}
/** {@inheritDoc} */
- @Override public void addListener(ConfigurationStorageListener listener) {
+ @Override public synchronized void addListener(ConfigurationStorageListener listener) {
listeners.add(listener);
+
+ if (watchId == null) {
+ // TODO: registerWatchByPrefix could throw OperationTimeoutException and CompactedException and we should
+ // TODO: properly handle such cases https://issues.apache.org/jira/browse/IGNITE-14604
+ watchId = metaStorageMgr.registerWatchByPrefix(MASTER_KEY, new WatchListener() {
+ @Override public boolean onUpdate(@NotNull Iterable<WatchEvent> events) {
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ long maxRevision = 0L;
+
+ Entry entryForMasterKey = null;
+
+ for (WatchEvent event : events) {
+ Entry e = event.newEntry();
+
+ if (!e.key().equals(MASTER_KEY)) {
+ data.put(e.key().toString().substring((DISTRIBUTED_PREFIX).length()),
+ (Serializable)ByteUtils.fromBytes(e.value()));
+
+ if (maxRevision < e.revision())
+ maxRevision = e.revision();
+ } else
+ entryForMasterKey = e;
+ }
+
+ // Contract of metastorage ensures that all updates of one revision will come in one batch.
+ // Also masterKey should be updated every time when we update cfg.
+ // That means that masterKey update must be included in the batch.
+ assert entryForMasterKey != null;
+
+ assert maxRevision == entryForMasterKey.revision();
+
+ assert maxRevision >= ver.get();
+
+ long finalMaxRevision = maxRevision;
+
+ listeners.forEach(listener -> listener.onEntriesChanged(new Data(data, finalMaxRevision)));
+
+ return true;
+ }
+
+ @Override public void onError(@NotNull Throwable e) {
+ // TODO: need to handle this case and there should some mechanism for registering new watch as far as
+ // TODO: onError unregisters failed watch https://issues.apache.org/jira/browse/IGNITE-14604
+ LOG.error("Metastorage listener issue", e);
+ }
+ });
+ }
}
/** {@inheritDoc} */
- @Override public void removeListener(ConfigurationStorageListener listener) {
+ @Override public synchronized void removeListener(ConfigurationStorageListener listener) {
listeners.remove(listener);
+
+ if (listeners.isEmpty()) {
+ try {
+ metaStorageMgr.unregisterWatch(watchId.get());
+ }
+ catch (InterruptedException | ExecutionException e) {
+ LOG.error("Failed to register watch in metastore", e);
+ }
+
+ watchId = null;
+ }
}
/** {@inheritDoc} */
@Override public void notifyApplied(long storageRevision) {
- // No-op.
+ assert ver.get() <= storageRevision;
+
+ ver.set(storageRevision);
+
+ // TODO: Also we should persist version,
+ // TODO: this should be done when nodes restart is introduced.
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-14697
}
/** {@inheritDoc} */
@Override public ConfigurationType type() {
return ConfigurationType.DISTRIBUTED;
}
+
+ /**
+ * Method that returns all distributed configuration keys from metastorage filtered out by the current applied
+ * revision as an upper bound. Applied revision is a revision of the last successful vault update.
+ * <p>
+ * This is possible to distinguish cfg keys from metastorage because we add special prefix {@link
+ * DistributedConfigurationStorage#DISTRIBUTED_PREFIX} to all configuration keys that we put to metastorage.
+ *
+ * @return Cursor built upon all distributed configuration entries.
+ */
+ private Cursor<Entry> allDstCfgKeys() {
Review comment:
Please do not use abbreviations for method names.
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/DistributedConfigurationStorage.java
##########
@@ -18,25 +18,60 @@
package org.apache.ignite.internal.storage;
import java.io.Serializable;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.configuration.storage.ConfigurationStorage;
import org.apache.ignite.configuration.storage.ConfigurationStorageListener;
import org.apache.ignite.configuration.storage.ConfigurationType;
import org.apache.ignite.configuration.storage.Data;
import org.apache.ignite.configuration.storage.StorageException;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.lang.IgniteLogger;
+import org.apache.ignite.metastorage.common.Conditions;
+import org.apache.ignite.metastorage.common.Cursor;
+import org.apache.ignite.metastorage.common.Entry;
+import org.apache.ignite.metastorage.common.Key;
+import org.apache.ignite.metastorage.common.Operation;
+import org.apache.ignite.metastorage.common.Operations;
+import org.apache.ignite.metastorage.common.WatchEvent;
+import org.apache.ignite.metastorage.common.WatchListener;
+import org.jetbrains.annotations.NotNull;
/**
* Distributed configuration storage.
*/
-// TODO: IGNITE-14586 Remove @SuppressWarnings when implementation provided.
-@SuppressWarnings({"FieldCanBeLocal", "unused"}) public class DistributedConfigurationStorage implements ConfigurationStorage {
+public class DistributedConfigurationStorage implements ConfigurationStorage {
+ /** Prefix that we add to configuration keys to distinguish them in metastorage. Must end with dot. */
+ private static final String DISTRIBUTED_PREFIX = "dst-cfg.";
+
+ /** Logger. */
+ private static final IgniteLogger LOG = IgniteLogger.forClass(DistributedConfigurationStorage.class);
+
+ /**
+ * Key for CAS-ing configuration keys to metastorage. This key is expected to be the first key in lexicographical
+ * order of distributed configuration keys.
+ */
+ private static final Key MASTER_KEY = new Key(DISTRIBUTED_PREFIX);
+
+ /**
+ * This key is expected to be the last key in lexicographical order of distributed configuration keys. It is
+ * possible because keys are in lexicographical order in metastorage and adding {@code (char)('.' + 1)} to the end
+ * will produce all keys with prefix {@link DistributedConfigurationStorage#DISTRIBUTED_PREFIX}
+ */
+ private static final Key dstKeysEndRange = new Key(DISTRIBUTED_PREFIX.substring(0, DISTRIBUTED_PREFIX.length() - 1) + (char)('.' + 1));
Review comment:
static final fields must be upper-cases
https://cwiki.apache.org/confluence/display/IGNITE/Coding+Guidelines#CodingGuidelines-Naming
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/LocalConfigurationStorage.java
##########
@@ -49,52 +54,72 @@ public LocalConfigurationStorage(VaultManager vaultMgr) {
this.vaultMgr = vaultMgr;
}
- /** Map to store values. */
- private Map<String, Serializable> map = new ConcurrentHashMap<>();
-
/** Change listeners. */
private List<ConfigurationStorageListener> listeners = new CopyOnWriteArrayList<>();
/** Storage version. */
- private AtomicLong version = new AtomicLong(0);
+ private AtomicLong ver = new AtomicLong(0);
+
+ /** Start key in range for searching local configuration keys. */
+ private static final ByteArray LOC_KEYS_START_RANGE = ByteArray.fromString(LOC_PREFIX);
+
+ /** End key in range for searching local configuration keys. */
+ private static final ByteArray LOC_KEYS_END_RANGE = ByteArray.fromString(LOC_PREFIX.substring(0, LOC_PREFIX.length() - 1) + (char)('.' + 1));
/** {@inheritDoc} */
@Override public synchronized Data readAll() throws StorageException {
- return new Data(new HashMap<>(map), version.get(), 0);
+ Iterator<Entry> iter =
+ vaultMgr.range(LOC_KEYS_START_RANGE, LOC_KEYS_END_RANGE);
+
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ while (iter.hasNext()) {
+ Entry val = iter.next();
+
+ data.put(val.key().toString().substring(LOC_KEYS_START_RANGE.toString().length()),
+ (Serializable)ByteUtils.fromBytes(val.value()));
+ }
+
+ // TODO: Need to restore version from pds when restart will be developed
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-14697
+ return new Data(data, ver.get());
}
/** {@inheritDoc} */
- @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long version) {
- if (version != this.version.get())
+ @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long sentVersion) {
+ if (sentVersion != ver.get())
return CompletableFuture.completedFuture(false);
- for (Map.Entry<String, Serializable> entry : newValues.entrySet()) {
- if (entry.getValue() != null)
- map.put(entry.getKey(), entry.getValue());
- else
- map.remove(entry.getKey());
- }
+ Map<ByteArray, byte[]> data = new HashMap<>();
+
+ for (Map.Entry<String, Serializable> e: newValues.entrySet()) {
+ ByteArray key = ByteArray.fromString(LOC_PREFIX + e.getKey());
- this.version.incrementAndGet();
+ data.put(key, e.getValue() == null ? null : ByteUtils.toBytes(e.getValue()));
+ }
- listeners.forEach(listener -> listener.onEntriesChanged(new Data(newValues, this.version.get(), 0)));
+ return vaultMgr.putAll(data).thenApply(res -> {
+ listeners.forEach(listener -> listener.onEntriesChanged(new Data(newValues, ver.incrementAndGet())));
- return CompletableFuture.completedFuture(true);
+ return true;
+ });
}
/** {@inheritDoc} */
- @Override public void addListener(ConfigurationStorageListener listener) {
- listeners.add(listener);
+ @Override public synchronized void addListener(ConfigurationStorageListener lsnr) {
Review comment:
Do we really need `synchronized here?
##########
File path: modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultManager.java
##########
@@ -105,13 +106,27 @@ public boolean bootstrapped() {
}
/**
- * Inserts or updates entries with given keys and given values and non-negative revision.
+ * Inserts or updates entries with given keys and given values. If the given value in {@code vals} is null,
Review comment:
here and below `{@code null}`
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/DistributedConfigurationStorage.java
##########
@@ -49,56 +84,178 @@ public DistributedConfigurationStorage(MetaStorageManager metaStorageMgr) {
this.metaStorageMgr = metaStorageMgr;
}
- /** Map to store values. */
- private Map<String, Serializable> map = new ConcurrentHashMap<>();
-
/** Change listeners. */
private List<ConfigurationStorageListener> listeners = new CopyOnWriteArrayList<>();
- /** Storage version. */
- private AtomicLong version = new AtomicLong(0);
+ /** Storage version. It stores actual metastorage revision, that is applied to configuration manager. */
Review comment:
Please define all fields before methods and constructors.
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/DistributedConfigurationStorage.java
##########
@@ -49,56 +84,178 @@ public DistributedConfigurationStorage(MetaStorageManager metaStorageMgr) {
this.metaStorageMgr = metaStorageMgr;
}
- /** Map to store values. */
- private Map<String, Serializable> map = new ConcurrentHashMap<>();
-
/** Change listeners. */
private List<ConfigurationStorageListener> listeners = new CopyOnWriteArrayList<>();
- /** Storage version. */
- private AtomicLong version = new AtomicLong(0);
+ /** Storage version. It stores actual metastorage revision, that is applied to configuration manager. */
+ private AtomicLong ver = new AtomicLong(0L);
/** {@inheritDoc} */
@Override public synchronized Data readAll() throws StorageException {
- return new Data(new HashMap<>(map), version.get(), 0);
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ Iterator<Entry> entries = allDstCfgKeys().iterator();
+
+ long maxRevision = 0L;
+
+ if (!entries.hasNext())
+ return new Data(data, ver.get());
+
+ Entry entryForMasterKey = entries.next();
+
+ // First key must be the masterKey because it's supposed to be the first in lexicographical order
+ assert entryForMasterKey.key().equals(MASTER_KEY);
+
+ while (entries.hasNext()) {
+ Entry entry = entries.next();
+
+ data.put(entry.key().toString().substring((DISTRIBUTED_PREFIX).length()), (Serializable)ByteUtils.fromBytes(entry.value()));
+
+ // Move to stream
+ if (maxRevision < entry.revision())
+ maxRevision = entry.revision();
+
+ }
+
+ if (!data.isEmpty()) {
+ assert maxRevision == entryForMasterKey.revision();
+
+ assert maxRevision >= ver.get();
+
+ return new Data(data, maxRevision);
+ }
+
+ return new Data(data, ver.get());
}
/** {@inheritDoc} */
- @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long version) {
- if (version != this.version.get())
+ @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long sentVersion) {
+ assert sentVersion <= ver.get();
+
+ if (sentVersion != ver.get())
+ // This means that sentVersion is less than version and other node has already updated configuration and
+ // write should be retried. Actual version will be set when watch and corresponding configuration listener
+ // updates configuration and notifyApplied is triggered afterwards.
return CompletableFuture.completedFuture(false);
+ HashSet<Operation> operations = new HashSet<>();
+
for (Map.Entry<String, Serializable> entry : newValues.entrySet()) {
+ Key key = new Key(DISTRIBUTED_PREFIX + entry.getKey());
+
if (entry.getValue() != null)
- map.put(entry.getKey(), entry.getValue());
+ // TODO: investigate overhead when serialize int, long, double, boolean, string, arrays of above
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-14698
+ operations.add(Operations.put(key, ByteUtils.toBytes(entry.getValue())));
else
- map.remove(entry.getKey());
+ operations.add(Operations.remove(key));
}
- this.version.incrementAndGet();
-
- listeners.forEach(listener -> listener.onEntriesChanged(new Data(newValues, this.version.get(), 0)));
+ operations.add(Operations.put(MASTER_KEY, ByteUtils.longToBytes(sentVersion)));
- return CompletableFuture.completedFuture(true);
+ return metaStorageMgr.invoke(
+ Conditions.key(MASTER_KEY).revision().eq(ver.get()),
+ operations,
+ Collections.singleton(Operations.noop()));
}
/** {@inheritDoc} */
- @Override public void addListener(ConfigurationStorageListener listener) {
+ @Override public synchronized void addListener(ConfigurationStorageListener listener) {
listeners.add(listener);
+
+ if (watchId == null) {
+ // TODO: registerWatchByPrefix could throw OperationTimeoutException and CompactedException and we should
+ // TODO: properly handle such cases https://issues.apache.org/jira/browse/IGNITE-14604
+ watchId = metaStorageMgr.registerWatchByPrefix(MASTER_KEY, new WatchListener() {
+ @Override public boolean onUpdate(@NotNull Iterable<WatchEvent> events) {
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ long maxRevision = 0L;
+
+ Entry entryForMasterKey = null;
+
+ for (WatchEvent event : events) {
+ Entry e = event.newEntry();
+
+ if (!e.key().equals(MASTER_KEY)) {
+ data.put(e.key().toString().substring((DISTRIBUTED_PREFIX).length()),
+ (Serializable)ByteUtils.fromBytes(e.value()));
+
+ if (maxRevision < e.revision())
+ maxRevision = e.revision();
+ } else
+ entryForMasterKey = e;
+ }
+
+ // Contract of metastorage ensures that all updates of one revision will come in one batch.
+ // Also masterKey should be updated every time when we update cfg.
+ // That means that masterKey update must be included in the batch.
+ assert entryForMasterKey != null;
+
+ assert maxRevision == entryForMasterKey.revision();
+
+ assert maxRevision >= ver.get();
+
+ long finalMaxRevision = maxRevision;
+
+ listeners.forEach(listener -> listener.onEntriesChanged(new Data(data, finalMaxRevision)));
+
+ return true;
+ }
+
+ @Override public void onError(@NotNull Throwable e) {
+ // TODO: need to handle this case and there should some mechanism for registering new watch as far as
+ // TODO: onError unregisters failed watch https://issues.apache.org/jira/browse/IGNITE-14604
+ LOG.error("Metastorage listener issue", e);
+ }
+ });
+ }
}
/** {@inheritDoc} */
- @Override public void removeListener(ConfigurationStorageListener listener) {
+ @Override public synchronized void removeListener(ConfigurationStorageListener listener) {
listeners.remove(listener);
+
+ if (listeners.isEmpty()) {
+ try {
+ metaStorageMgr.unregisterWatch(watchId.get());
+ }
+ catch (InterruptedException | ExecutionException e) {
+ LOG.error("Failed to register watch in metastore", e);
Review comment:
`Failed to unregister watch in meta storage.`
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/LocalConfigurationStorage.java
##########
@@ -49,52 +54,72 @@ public LocalConfigurationStorage(VaultManager vaultMgr) {
this.vaultMgr = vaultMgr;
}
- /** Map to store values. */
- private Map<String, Serializable> map = new ConcurrentHashMap<>();
-
/** Change listeners. */
private List<ConfigurationStorageListener> listeners = new CopyOnWriteArrayList<>();
/** Storage version. */
- private AtomicLong version = new AtomicLong(0);
+ private AtomicLong ver = new AtomicLong(0);
Review comment:
Please move all field definitions before the constructor and other methods.
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/LocalConfigurationStorage.java
##########
@@ -49,52 +54,72 @@ public LocalConfigurationStorage(VaultManager vaultMgr) {
this.vaultMgr = vaultMgr;
}
- /** Map to store values. */
- private Map<String, Serializable> map = new ConcurrentHashMap<>();
-
/** Change listeners. */
private List<ConfigurationStorageListener> listeners = new CopyOnWriteArrayList<>();
/** Storage version. */
- private AtomicLong version = new AtomicLong(0);
+ private AtomicLong ver = new AtomicLong(0);
+
+ /** Start key in range for searching local configuration keys. */
+ private static final ByteArray LOC_KEYS_START_RANGE = ByteArray.fromString(LOC_PREFIX);
+
+ /** End key in range for searching local configuration keys. */
+ private static final ByteArray LOC_KEYS_END_RANGE = ByteArray.fromString(LOC_PREFIX.substring(0, LOC_PREFIX.length() - 1) + (char)('.' + 1));
/** {@inheritDoc} */
@Override public synchronized Data readAll() throws StorageException {
- return new Data(new HashMap<>(map), version.get(), 0);
+ Iterator<Entry> iter =
+ vaultMgr.range(LOC_KEYS_START_RANGE, LOC_KEYS_END_RANGE);
+
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ while (iter.hasNext()) {
+ Entry val = iter.next();
+
+ data.put(val.key().toString().substring(LOC_KEYS_START_RANGE.toString().length()),
+ (Serializable)ByteUtils.fromBytes(val.value()));
+ }
+
+ // TODO: Need to restore version from pds when restart will be developed
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-14697
+ return new Data(data, ver.get());
}
/** {@inheritDoc} */
- @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long version) {
- if (version != this.version.get())
+ @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long sentVersion) {
+ if (sentVersion != ver.get())
return CompletableFuture.completedFuture(false);
- for (Map.Entry<String, Serializable> entry : newValues.entrySet()) {
- if (entry.getValue() != null)
- map.put(entry.getKey(), entry.getValue());
- else
- map.remove(entry.getKey());
- }
+ Map<ByteArray, byte[]> data = new HashMap<>();
+
+ for (Map.Entry<String, Serializable> e: newValues.entrySet()) {
+ ByteArray key = ByteArray.fromString(LOC_PREFIX + e.getKey());
- this.version.incrementAndGet();
+ data.put(key, e.getValue() == null ? null : ByteUtils.toBytes(e.getValue()));
+ }
- listeners.forEach(listener -> listener.onEntriesChanged(new Data(newValues, this.version.get(), 0)));
+ return vaultMgr.putAll(data).thenApply(res -> {
+ listeners.forEach(listener -> listener.onEntriesChanged(new Data(newValues, ver.incrementAndGet())));
Review comment:
It means that everyone listener will be updated with a new `Data` object with a different revision. Looks weird to me. Am I missing something?
##########
File path: modules/runner/src/main/java/org/apache/ignite/internal/storage/DistributedConfigurationStorage.java
##########
@@ -49,56 +84,178 @@ public DistributedConfigurationStorage(MetaStorageManager metaStorageMgr) {
this.metaStorageMgr = metaStorageMgr;
}
- /** Map to store values. */
- private Map<String, Serializable> map = new ConcurrentHashMap<>();
-
/** Change listeners. */
private List<ConfigurationStorageListener> listeners = new CopyOnWriteArrayList<>();
- /** Storage version. */
- private AtomicLong version = new AtomicLong(0);
+ /** Storage version. It stores actual metastorage revision, that is applied to configuration manager. */
+ private AtomicLong ver = new AtomicLong(0L);
/** {@inheritDoc} */
@Override public synchronized Data readAll() throws StorageException {
- return new Data(new HashMap<>(map), version.get(), 0);
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ Iterator<Entry> entries = allDstCfgKeys().iterator();
+
+ long maxRevision = 0L;
+
+ if (!entries.hasNext())
+ return new Data(data, ver.get());
+
+ Entry entryForMasterKey = entries.next();
+
+ // First key must be the masterKey because it's supposed to be the first in lexicographical order
+ assert entryForMasterKey.key().equals(MASTER_KEY);
+
+ while (entries.hasNext()) {
+ Entry entry = entries.next();
+
+ data.put(entry.key().toString().substring((DISTRIBUTED_PREFIX).length()), (Serializable)ByteUtils.fromBytes(entry.value()));
+
+ // Move to stream
+ if (maxRevision < entry.revision())
+ maxRevision = entry.revision();
+
+ }
+
+ if (!data.isEmpty()) {
+ assert maxRevision == entryForMasterKey.revision();
+
+ assert maxRevision >= ver.get();
+
+ return new Data(data, maxRevision);
+ }
+
+ return new Data(data, ver.get());
}
/** {@inheritDoc} */
- @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long version) {
- if (version != this.version.get())
+ @Override public synchronized CompletableFuture<Boolean> write(Map<String, Serializable> newValues, long sentVersion) {
+ assert sentVersion <= ver.get();
+
+ if (sentVersion != ver.get())
+ // This means that sentVersion is less than version and other node has already updated configuration and
+ // write should be retried. Actual version will be set when watch and corresponding configuration listener
+ // updates configuration and notifyApplied is triggered afterwards.
return CompletableFuture.completedFuture(false);
+ HashSet<Operation> operations = new HashSet<>();
+
for (Map.Entry<String, Serializable> entry : newValues.entrySet()) {
+ Key key = new Key(DISTRIBUTED_PREFIX + entry.getKey());
+
if (entry.getValue() != null)
- map.put(entry.getKey(), entry.getValue());
+ // TODO: investigate overhead when serialize int, long, double, boolean, string, arrays of above
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-14698
+ operations.add(Operations.put(key, ByteUtils.toBytes(entry.getValue())));
else
- map.remove(entry.getKey());
+ operations.add(Operations.remove(key));
}
- this.version.incrementAndGet();
-
- listeners.forEach(listener -> listener.onEntriesChanged(new Data(newValues, this.version.get(), 0)));
+ operations.add(Operations.put(MASTER_KEY, ByteUtils.longToBytes(sentVersion)));
- return CompletableFuture.completedFuture(true);
+ return metaStorageMgr.invoke(
+ Conditions.key(MASTER_KEY).revision().eq(ver.get()),
Review comment:
The `ver` variable can be simultaneously updated via `notifyApplied` method. Perhaps, it makes sense to read this variable only once at the very beginning of `write` method and use that value everywhere. What do you think?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org