alievmirza commented on a change in pull request #114:
URL: https://github.com/apache/ignite-3/pull/114#discussion_r631010321
##########
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:
done
--
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]
