-
Notifications
You must be signed in to change notification settings - Fork 1.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
CURATOR-688. SharedCount will be never updated successful when version of ZNode is overflow. #478
Changes from all commits
ff801a8
174a8a1
849eaa4
d333087
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,39 @@ | ||
/* | ||
* 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.curator.framework.recipes.shared; | ||
|
||
import org.apache.zookeeper.data.Stat; | ||
|
||
/** | ||
* Exception to alert overflowed {@link Stat#getVersion()} {@code -1} which is not suitable in | ||
* {@link SharedValue#trySetValue(VersionedValue, byte[])} and {@link SharedCount#trySetCount(VersionedValue, int)}. | ||
* | ||
* <p>In case of this exception, clients have to choose: | ||
* <ul> | ||
* <li>Take their own risk to do a blind set.</li> | ||
* <li>Update ZooKeeper cluster to solve <a href="https://issues.apache.org/jira/browse/ZOOKEEPER-4743">ZOOKEEPER-4743</a>.</li> | ||
* </ul> | ||
*/ | ||
public class IllegalTrySetVersionException extends IllegalArgumentException { | ||
@Override | ||
public String getMessage() { | ||
return "overflowed Stat.version -1 is not suitable for trySet(a.k.a. compare-and-set ZooKeeper::setData)"; | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -48,6 +48,7 @@ | |
* value (considering ZK's normal consistency guarantees). | ||
*/ | ||
public class SharedValue implements Closeable, SharedValueReader { | ||
private static final int NO_ZXID = -1; | ||
private static final int UNINITIALIZED_VERSION = -1; | ||
|
||
private final Logger log = LoggerFactory.getLogger(getClass()); | ||
|
@@ -101,8 +102,8 @@ public SharedValue(CuratorFramework client, String path, byte[] seedValue) { | |
this.path = PathUtils.validatePath(path); | ||
this.seedValue = Arrays.copyOf(seedValue, seedValue.length); | ||
this.watcher = new SharedValueCuratorWatcher(); | ||
currentValue = new AtomicReference<VersionedValue<byte[]>>( | ||
new VersionedValue<byte[]>(UNINITIALIZED_VERSION, Arrays.copyOf(seedValue, seedValue.length))); | ||
currentValue = new AtomicReference<>( | ||
new VersionedValue<>(NO_ZXID, UNINITIALIZED_VERSION, Arrays.copyOf(seedValue, seedValue.length))); | ||
} | ||
|
||
@VisibleForTesting | ||
|
@@ -112,8 +113,8 @@ protected SharedValue(WatcherRemoveCuratorFramework client, String path, byte[] | |
this.seedValue = Arrays.copyOf(seedValue, seedValue.length); | ||
// inject watcher for testing | ||
this.watcher = watcher; | ||
currentValue = new AtomicReference<VersionedValue<byte[]>>( | ||
new VersionedValue<byte[]>(UNINITIALIZED_VERSION, Arrays.copyOf(seedValue, seedValue.length))); | ||
currentValue = new AtomicReference<>( | ||
new VersionedValue<>(NO_ZXID, UNINITIALIZED_VERSION, Arrays.copyOf(seedValue, seedValue.length))); | ||
} | ||
|
||
@Override | ||
|
@@ -125,12 +126,11 @@ public byte[] getValue() { | |
@Override | ||
public VersionedValue<byte[]> getVersionedValue() { | ||
VersionedValue<byte[]> localCopy = currentValue.get(); | ||
return new VersionedValue<byte[]>( | ||
localCopy.getVersion(), Arrays.copyOf(localCopy.getValue(), localCopy.getValue().length)); | ||
return localCopy.mapValue(bytes -> Arrays.copyOf(bytes, bytes.length)); | ||
} | ||
|
||
/** | ||
* Change the shared value value irrespective of its previous state | ||
* Change the shared value irrespective of its previous state | ||
* | ||
* @param newValue new value | ||
* @throws Exception ZK errors, interruptions, etc. | ||
|
@@ -139,7 +139,7 @@ public void setValue(byte[] newValue) throws Exception { | |
Preconditions.checkState(state.get() == State.STARTED, "not started"); | ||
|
||
Stat result = client.setData().forPath(path, newValue); | ||
updateValue(result.getVersion(), Arrays.copyOf(newValue, newValue.length)); | ||
updateValue(result.getMzxid(), result.getVersion(), Arrays.copyOf(newValue, newValue.length)); | ||
} | ||
|
||
/** | ||
|
@@ -171,19 +171,25 @@ public boolean trySetValue(byte[] newValue) throws Exception { | |
* @param newValue the new value to attempt | ||
* @return true if the change attempt was successful, false if not. If the change | ||
* was not successful, {@link #getValue()} will return the updated value | ||
* @throws IllegalTrySetVersionException if {@link Stat#getVersion()} overflowed to {@code -1} | ||
* @throws Exception ZK errors, interruptions, etc. | ||
*/ | ||
public boolean trySetValue(VersionedValue<byte[]> previous, byte[] newValue) throws Exception { | ||
Preconditions.checkState(state.get() == State.STARTED, "not started"); | ||
|
||
VersionedValue<byte[]> current = currentValue.get(); | ||
if (previous.getVersion() != current.getVersion() || !Arrays.equals(previous.getValue(), current.getValue())) { | ||
// Omit comparing of getVersion here, so we can test the exception case. | ||
// This affects no correctness as construction of VersionedValue is private. | ||
if (previous.getZxid() != current.getZxid() || !Arrays.equals(previous.getValue(), current.getValue())) { | ||
return false; | ||
} | ||
if (previous.getVersion() == -1) { | ||
throw new IllegalTrySetVersionException(); | ||
} | ||
|
||
try { | ||
Stat result = client.setData().withVersion(previous.getVersion()).forPath(path, newValue); | ||
updateValue(result.getVersion(), Arrays.copyOf(newValue, newValue.length)); | ||
updateValue(result.getMzxid(), result.getVersion(), Arrays.copyOf(newValue, newValue.length)); | ||
return true; | ||
} catch (KeeperException.BadVersionException ignore) { | ||
// ignore | ||
|
@@ -193,14 +199,13 @@ public boolean trySetValue(VersionedValue<byte[]> previous, byte[] newValue) thr | |
return false; | ||
} | ||
|
||
private void updateValue(int version, byte[] bytes) { | ||
private void updateValue(long zxid, int version, byte[] bytes) { | ||
while (true) { | ||
VersionedValue<byte[]> current = currentValue.get(); | ||
if (current.getVersion() >= version) { | ||
// A newer version was concurrently set. | ||
if (current.getZxid() >= zxid) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. What if only one client here and it goes through the overflow bound? Said Or we have different assumption on zxid? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is paranoid -:).
In case of above situation, I believed that ZooKeeper is doomed to failure. The "never updated" should be negligible in case of the disaster. |
||
return; | ||
} | ||
if (currentValue.compareAndSet(current, new VersionedValue<byte[]>(version, bytes))) { | ||
if (currentValue.compareAndSet(current, new VersionedValue<>(zxid, version, bytes))) { | ||
// Successfully set. | ||
return; | ||
} | ||
|
@@ -248,14 +253,14 @@ private void readValue() throws Exception { | |
Stat localStat = new Stat(); | ||
byte[] bytes = | ||
client.getData().storingStatIn(localStat).usingWatcher(watcher).forPath(path); | ||
updateValue(localStat.getVersion(), bytes); | ||
updateValue(localStat.getMzxid(), localStat.getVersion(), bytes); | ||
} | ||
|
||
private final BackgroundCallback upadateAndNotifyListenerCallback = new BackgroundCallback() { | ||
@Override | ||
public void processResult(CuratorFramework client, CuratorEvent event) throws Exception { | ||
if (event.getResultCode() == KeeperException.Code.OK.intValue()) { | ||
updateValue(event.getStat().getVersion(), event.getData()); | ||
updateValue(event.getStat().getMzxid(), event.getStat().getVersion(), event.getData()); | ||
notifyListeners(); | ||
} | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I am not sure wether
SharedValue
was designed to work with multiple owners, but I saw there is a background watcher to update value and also there is no rule to forbid concurrent usages. So, I assume it should work well in case of concurrency.Then, let me assume a situation:
current.getVersion
isInteger.MAX_VALUE
.trySetValue
and succeed to get overflowed versionInteger.MIN_VALUE
, but the call toupdateValue
is somewhat delayed.updateVersion
with versionInteger.MIN_VALUE + 1
. According to the code, this will be ignored.updateValue
to continue its task with versionInteger.MIN_VALUE
. It succeeds.current
stores dated version while the javadoc says "All clients watching the same path will have the up-to-date value (considering ZK's normal consistency guarantees)".There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I am not convinced in the overflow case either. @tisonkun @Hexiaoqiao
For the overflow case,
VersionedValue#getVersion
so to warn clients about the "ordering assumption" if any aboutversion
.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we should refactor
updateValue
a bit to compare ordering usingStat.mzxid
. This way we are not fearing this overflow issue. I am stupid in reviwing without a deep thought, sorry for that. So my finally points are:VersionedValue#getVersion
to warn clients about "ordering assumptions" and "overflow behavior".updateValue
to order usingStat.mzxid
.-1
Stat.version
intrySetValue
. I am positive to ZOOKEEPER-4743.trySetValue
.Besides above, should we expose a
VersionedValue#getZxid
for client usage ?Any thoughts @tisonkun @eolivelli @Hexiaoqiao ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I have pushed new commits to go through above direction. Could you please take a look @Hexiaoqiao @tisonkun @eolivelli ?