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:
[email protected]
