sanpwc commented on a change in pull request #114:
URL: https://github.com/apache/ignite-3/pull/114#discussion_r627600834
##########
File path:
modules/runner/src/main/java/org/apache/ignite/internal/storage/DistributedConfigurationStorage.java
##########
@@ -49,52 +72,154 @@ 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 version = new AtomicLong(0L);
/** {@inheritDoc} */
@Override public synchronized Data readAll() throws StorageException {
- return new Data(new HashMap<>(map), version.get(), 0);
+
+ Cursor<Entry> cur = metaStorageMgr.rangeWithAppliedRevision(new
Key(DISTRIBUTED_PREFIX + "."),
+ new Key(DISTRIBUTED_PREFIX + (char)('.' + 1)));
+
+ HashMap<String, Serializable> data = new HashMap<>();
+
+ long maxRevision = 0L;
+
+ Entry entryForMasterKey = null;
+
+ for (Entry entry : cur) {
+ if (!entry.key().equals(masterKey)) {
+
data.put(entry.key().toString().replaceFirst(DISTRIBUTED_PREFIX + ".", ""),
+ (Serializable)ByteUtils.fromBytes(entry.value()));
+
+ // Move to stream
+ if (maxRevision < entry.revision())
+ maxRevision = entry.revision();
+ } else
+ entryForMasterKey = entry;
+ }
+
+ if (!data.isEmpty()) {
+ assert entryForMasterKey != null;
+
+ assert maxRevision == entryForMasterKey.revision();
+
+ assert maxRevision >= version.get();
+
+ return new Data(data, maxRevision);
+ }
+
+ return new Data(data, version.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 <= version.get();
+
+ if (sentVersion != version.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<>();
+
+ HashSet<Operation> failures = 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 deserialize int, long,
double, boolean, string, arrays of above
+ operations.add(Operations.put(key,
ByteUtils.toBytes(entry.getValue())));
else
- map.remove(entry.getKey());
- }
+ operations.add(Operations.remove(key));
- this.version.incrementAndGet();
+ failures.add(Operations.noop());
Review comment:
Is it expected by invoke() contract to have same amount of failure items
as a success ones? If it's true, could you please explain why it's required. If
it's false, let's just use
`return
metaStorageMgr.invoke(Conditions.key(masterKey).revision().eq(version.get()),
Operations.noop());`
instead of
`return
metaStorageMgr.invoke(Conditions.key(masterKey).revision().eq(version.get()),
operations, failures);`
--
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]
