murblanc commented on a change in pull request #2291:
URL: https://github.com/apache/lucene-solr/pull/2291#discussion_r568912229
##########
File path:
solr/core/src/java/org/apache/solr/cluster/events/impl/DelegatingClusterEventProducer.java
##########
@@ -90,7 +96,9 @@ public void setDelegate(ClusterEventProducer newDelegate) {
log.debug("--- delegate {} already in state {}", delegate,
delegate.getState());
}
}
- this.version++;
+ if (versionTracker != null) {
Review comment:
We have a synchronization issue (memory barrier type, not concurrent
access type). The thread calling `setDelegate()` is accessing `versionTracker`
set by another thread without synchronization.
Can be fixed by making `versionTracker` volatile.
##########
File path: solr/core/src/test/org/apache/solr/cluster/VersionTrackerImpl.java
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.solr.cluster;
+
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.util.TimeOut;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class VersionTrackerImpl implements VersionTracker {
+ private int version = 0;
+
+ @Override
+ public synchronized void increment() {
+ version++;
+ this.notifyAll();
+ }
+
+ @Override
+ public int waitForVersionChange(int currentVersion, int timeoutSec) throws
InterruptedException, TimeoutException {
+ TimeOut timeout = new TimeOut(timeoutSec, TimeUnit.SECONDS,
TimeSource.NANO_TIME);
+ int newVersion = currentVersion;
+ while (! timeout.hasTimedOut()) {
+ synchronized (this) {
+ if ((newVersion = version) != currentVersion) {
+ break;
+ }
+ this.wait(timeout.timeLeft(TimeUnit.MILLISECONDS));
+ }
+ }
+ if (newVersion < currentVersion) {
+ // ArithmeticException? This means we overflowed
+ throw new RuntimeException("Invalid version - went back!
currentVersion=" + currentVersion +
+ " newVersion=" + newVersion);
+ } else if (newVersion == currentVersion) {
+ throw new TimeoutException("Timed out waiting for version
change.");
Review comment:
Add the version value to the exception, might help debug tests.
##########
File path: solr/core/src/java/org/apache/solr/cluster/VersionTracker.java
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.solr.cluster;
+
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Allows for tracking state change from test classes. Typical use will be to
set a version tracker on a stateful
+ * object, which will call {@link #increment()} every time state changes. Test
clients observing the state will call
+ * {@link #waitForVersionChange(int, int)} to be notified of the next
increment call.
+ */
+public interface VersionTracker {
Review comment:
Tracking versions by incrementing is one possible implementation of this
interface, but maybe the interface doesn't have to hint that this is the
implementation?
Renaming `increment` into `notifyEvent` or something similar and
`VersionTracker` into `NotificationCallback` would make it more generic (not
suggesting these actual names, but you get the idea).
`waitForVersionChange` doesn't have to be part of the interface. Test code
can use the implementation it instantiates directly. Some tests might not even
want to wait but rather do other things with the notification (like count how
many notifications are received, or make sure none are received etc).
##########
File path: solr/core/src/test/org/apache/solr/cluster/VersionTrackerImpl.java
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.solr.cluster;
+
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.util.TimeOut;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class VersionTrackerImpl implements VersionTracker {
+ private int version = 0;
+
+ @Override
+ public synchronized void increment() {
+ version++;
+ this.notifyAll();
+ }
+
+ @Override
+ public int waitForVersionChange(int currentVersion, int timeoutSec) throws
InterruptedException, TimeoutException {
+ TimeOut timeout = new TimeOut(timeoutSec, TimeUnit.SECONDS,
TimeSource.NANO_TIME);
+ int newVersion = currentVersion;
+ while (! timeout.hasTimedOut()) {
Review comment:
formatting: remove space after `!`
##########
File path:
solr/core/src/java/org/apache/solr/cluster/placement/impl/DelegatingPlacementPluginFactory.java
##########
@@ -39,18 +39,20 @@ public PlacementPlugin createPluginInstance() {
}
}
+ @VisibleForTesting
+ public void setVersionTracker(VersionTracker tracker) {
+ versionTracker = tracker;
+ }
+
public void setDelegate(PlacementPluginFactory<? extends
PlacementPluginConfig> delegate) {
this.delegate = delegate;
- this.version++;
+ if (versionTracker != null) {
Review comment:
If we want to be strict about correctness, we should copy
`versionTracker` into another variable then test and call, otherwise it can be
non `null` when evaluating the if condition then `null` on the next line when
dereferencing it.
##########
File path: solr/core/src/test/org/apache/solr/cluster/VersionTrackerImpl.java
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.solr.cluster;
+
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.util.TimeOut;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class VersionTrackerImpl implements VersionTracker {
+ private int version = 0;
+
+ @Override
+ public synchronized void increment() {
+ version++;
+ this.notifyAll();
+ }
+
+ @Override
+ public int waitForVersionChange(int currentVersion, int timeoutSec) throws
InterruptedException, TimeoutException {
+ TimeOut timeout = new TimeOut(timeoutSec, TimeUnit.SECONDS,
TimeSource.NANO_TIME);
+ int newVersion = currentVersion;
+ while (! timeout.hasTimedOut()) {
+ synchronized (this) {
+ if ((newVersion = version) != currentVersion) {
+ break;
+ }
+ this.wait(timeout.timeLeft(TimeUnit.MILLISECONDS));
+ }
+ }
+ if (newVersion < currentVersion) {
+ // ArithmeticException? This means we overflowed
Review comment:
Likely bad call from the test...
##########
File path: solr/core/src/test/org/apache/solr/cluster/VersionTrackerImpl.java
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.solr.cluster;
+
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.util.TimeOut;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class VersionTrackerImpl implements VersionTracker {
+ private int version = 0;
+
+ @Override
+ public synchronized void increment() {
+ version++;
+ this.notifyAll();
+ }
+
+ @Override
+ public int waitForVersionChange(int currentVersion, int timeoutSec) throws
InterruptedException, TimeoutException {
+ TimeOut timeout = new TimeOut(timeoutSec, TimeUnit.SECONDS,
TimeSource.NANO_TIME);
+ int newVersion = currentVersion;
+ while (! timeout.hasTimedOut()) {
Review comment:
I also realize I would have put the while loop inside the synchronized
block, and now I wonder which way is better... Re-acquiring it over and over
again or not holding it while seeing if a timeout occurred.
##########
File path:
solr/core/src/test/org/apache/solr/cluster/events/ClusterEventProducerTest.java
##########
@@ -102,7 +105,7 @@ public void teardown() throws Exception {
@Test
public void testEvents() throws Exception {
- int version = waitForVersionChange(-1, 10);
+ int version = versionTracker.waitForVersionChange(-1, 10);
Review comment:
Maybe we need a `getCurrentVersion` in `VersionTrackerImpl` so a given
instance can be reused more easily in a given test?
##########
File path:
solr/core/src/test/org/apache/solr/cluster/events/ClusterEventProducerTest.java
##########
@@ -102,7 +105,7 @@ public void teardown() throws Exception {
@Test
public void testEvents() throws Exception {
- int version = waitForVersionChange(-1, 10);
+ int version = versionTracker.waitForVersionChange(-1, 10);
Review comment:
Version starts at 0 in the implementation of the version tracker, so
what does it mean to wait for it to be -1?
##########
File path:
solr/core/src/test/org/apache/solr/cluster/CountingStateChangeListener.java
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.solr.cluster;
+
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.util.TimeOut;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * A counting StateChangeListener that will internally track how many times
{@link #stateChanged()} has been called.
+ * Consumers can compare the number of state change calls before and after an
event to determine if they should proceed,
+ * made simple with {@link #waitForVersionChange(int, int)} method.
+ */
+public class CountingStateChangeListener implements StateChangeListener {
+ private int version = 0;
+
+ @Override
+ public synchronized void stateChanged() {
+ version++;
+ this.notifyAll();
+ }
+
+ /**
+ * Given a last known number of state changes, wait for additional changes
to come in. If no state changes have
+ * occurred beyond the known value, this method will wait for additional
changes to come in.
+ * If the current number of change events is unknown to the caller, then
this method can be called with <tt>-1</tt>
+ * to return immediately with the number of events up to this point.
+ * @param lastVersion the previous number of changes seen
+ * @param timeoutSec how long to wait for additional changes to occur
+ * @return the number of changes seen since initialization
+ */
+ public int waitForVersionChange(int lastVersion, int timeoutSec) throws
InterruptedException, TimeoutException {
+ TimeOut timeout = new TimeOut(timeoutSec, TimeUnit.SECONDS,
TimeSource.NANO_TIME);
+ int newVersion = lastVersion;
+ synchronized (this) {
+ while (!timeout.hasTimedOut() && (newVersion = version) !=
lastVersion) {
Review comment:
I don't get the condition here. Shouldn't we loop while `version ==
lastVersion` (so we exit the loop when it changes or when we time out) rather
than loop while they're different?
I suspect that the test without this improved lockstep synchronization was
passing always on your machine and it continues to pass for the same reason.
##########
File path:
solr/core/src/test/org/apache/solr/cluster/CountingStateChangeListener.java
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.solr.cluster;
+
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.util.TimeOut;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * A counting StateChangeListener that will internally track how many times
{@link #stateChanged()} has been called.
+ * Consumers can compare the number of state change calls before and after an
event to determine if they should proceed,
+ * made simple with {@link #waitForVersionChange(int, int)} method.
+ */
+public class CountingStateChangeListener implements StateChangeListener {
+ private int version = 0;
+
+ @Override
+ public synchronized void stateChanged() {
+ version++;
+ this.notifyAll();
+ }
+
+ /**
+ * Given a last known number of state changes, wait for additional changes
to come in. If no state changes have
+ * occurred beyond the known value, this method will wait for additional
changes to come in.
+ * If the current number of change events is unknown to the caller, then
this method can be called with <tt>-1</tt>
+ * to return immediately with the number of events up to this point.
+ * @param lastVersion the previous number of changes seen
+ * @param timeoutSec how long to wait for additional changes to occur
+ * @return the number of changes seen since initialization
+ */
+ public int waitForVersionChange(int lastVersion, int timeoutSec) throws
InterruptedException, TimeoutException {
+ TimeOut timeout = new TimeOut(timeoutSec, TimeUnit.SECONDS,
TimeSource.NANO_TIME);
+ int newVersion = lastVersion;
Review comment:
This assignment is not needed. I assume the compiler is unhappy if it is
just removed, but maybe doing the assignment to `newVersion` first before the
timeout check in the while condition makes the compiler happy?
----------------------------------------------------------------
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
