mirror of https://github.com/apache/jclouds.git
ConcurrentOpenSocketFinder cleanup
This commit is contained in:
Adrian Cole
parent
4d2124bada
commit
565a54164b
|
|
@ -1,14 +1,35 @@
|
|||
/**
|
||||
* Licensed to jclouds, Inc. (jclouds) under one or more
|
||||
* contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. jclouds 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.jclouds.compute.util;
|
||||
|
||||
import static com.google.common.base.Preconditions.checkNotNull;
|
||||
import static com.google.common.base.Preconditions.checkState;
|
||||
import static com.google.common.base.Throwables.propagate;
|
||||
import static com.google.common.collect.Iterables.concat;
|
||||
import static com.google.common.collect.Iterables.size;
|
||||
import static com.google.common.collect.Iterables.transform;
|
||||
import static com.google.common.util.concurrent.Atomics.newReference;
|
||||
import static com.google.common.util.concurrent.MoreExecutors.listeningDecorator;
|
||||
import static com.google.common.util.concurrent.MoreExecutors.sameThreadExecutor;
|
||||
import static org.jclouds.compute.config.ComputeServiceProperties.TIMEOUT_NODE_RUNNING;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
|
@ -25,18 +46,17 @@ import org.jclouds.logging.Logger;
|
|||
import org.jclouds.predicates.RetryablePredicate;
|
||||
import org.jclouds.predicates.SocketOpen;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import com.google.common.base.Function;
|
||||
import com.google.common.base.Predicate;
|
||||
import com.google.common.base.Throwables;
|
||||
import com.google.common.collect.FluentIterable;
|
||||
import com.google.common.collect.ImmutableSet;
|
||||
import com.google.common.net.HostAndPort;
|
||||
import com.google.common.util.concurrent.Atomics;
|
||||
import com.google.common.util.concurrent.ListenableFuture;
|
||||
import com.google.common.util.concurrent.ListeningExecutorService;
|
||||
import com.google.common.util.concurrent.MoreExecutors;
|
||||
import com.google.inject.Inject;
|
||||
|
||||
public class ConcurrentOpenSocketFinder implements OpenSocketFinder {
|
||||
public final class ConcurrentOpenSocketFinder implements OpenSocketFinder {
|
||||
|
||||
@Resource
|
||||
@Named(ComputeServiceConstants.COMPUTE_LOGGER)
|
||||
|
|
@ -47,24 +67,32 @@ public class ConcurrentOpenSocketFinder implements OpenSocketFinder {
|
|||
private final ListeningExecutorService executor;
|
||||
|
||||
@Inject
|
||||
public ConcurrentOpenSocketFinder(SocketOpen socketTester,
|
||||
@Named(TIMEOUT_NODE_RUNNING) final Predicate<AtomicReference<NodeMetadata>> nodeRunning,
|
||||
@VisibleForTesting
|
||||
ConcurrentOpenSocketFinder(SocketOpen socketTester,
|
||||
@Named(TIMEOUT_NODE_RUNNING) Predicate<AtomicReference<NodeMetadata>> nodeRunning,
|
||||
@Named(Constants.PROPERTY_USER_THREADS) ExecutorService userThreads) {
|
||||
this.socketTester = socketTester;
|
||||
this.nodeRunning = nodeRunning;
|
||||
this.executor = MoreExecutors.listeningDecorator(userThreads);
|
||||
this.socketTester =checkNotNull(socketTester, "socketTester");
|
||||
this.nodeRunning = checkNotNull(nodeRunning, "nodeRunning");
|
||||
this.executor = listeningDecorator(checkNotNull(userThreads, "userThreads"));
|
||||
}
|
||||
|
||||
public HostAndPort findOpenSocketOnNode(final NodeMetadata node, final int port,
|
||||
public HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port,
|
||||
long timeoutValue, TimeUnit timeUnits) {
|
||||
Iterable<String> hosts = checkNodeHasIps(node);
|
||||
Set<HostAndPort> sockets = toHostAndPorts(hosts, port);
|
||||
FluentIterable<String> hosts = checkNodeHasIps(node);
|
||||
ImmutableSet<HostAndPort> sockets = hosts.transform(new Function<String, HostAndPort>() {
|
||||
|
||||
@Override
|
||||
public HostAndPort apply(String from) {
|
||||
return HostAndPort.fromParts(from, port);
|
||||
}
|
||||
}).toImmutableSet();
|
||||
|
||||
// Specify a retry period of 1s, expressed in the same time units.
|
||||
long period = timeUnits.convert(1, TimeUnit.SECONDS);
|
||||
|
||||
// For storing the result; needed because predicate will just tell us true/false
|
||||
final AtomicReference<HostAndPort> result = Atomics.newReference();
|
||||
final AtomicReference<HostAndPort> result = newReference();
|
||||
final AtomicReference<NodeMetadata> nodeReference = newReference(node);
|
||||
|
||||
Predicate<Collection<HostAndPort>> concurrentOpenSocketFinder = new Predicate<Collection<HostAndPort>>() {
|
||||
|
||||
|
|
@ -75,8 +103,8 @@ public class ConcurrentOpenSocketFinder implements OpenSocketFinder {
|
|||
result.set(reachableSocket);
|
||||
return true;
|
||||
} else {
|
||||
if (nodeRunning != null && !nodeRunning.apply(new AtomicReference<NodeMetadata>(node))) {
|
||||
throw new IllegalStateException(String.format("Node %s is no longer running; aborting waiting for ip:port connection", node.getId()));
|
||||
if (!nodeRunning.apply(nodeReference)) {
|
||||
throw new IllegalStateException(String.format("Node %s is no longer running; aborting waiting for ip:port connection", nodeReference.get().getId()));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
|
@ -111,7 +139,7 @@ public class ConcurrentOpenSocketFinder implements OpenSocketFinder {
|
|||
* @throws InterruptedException
|
||||
*/
|
||||
private HostAndPort findOpenSocket(final Collection<HostAndPort> sockets) {
|
||||
final AtomicReference<HostAndPort> result = Atomics.newReference();
|
||||
final AtomicReference<HostAndPort> result = newReference();
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
final AtomicInteger completeCount = new AtomicInteger();
|
||||
|
||||
|
|
@ -141,32 +169,22 @@ public class ConcurrentOpenSocketFinder implements OpenSocketFinder {
|
|||
}
|
||||
}
|
||||
|
||||
}, MoreExecutors.sameThreadExecutor());
|
||||
}, sameThreadExecutor());
|
||||
}
|
||||
|
||||
try {
|
||||
latch.await();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw Throwables.propagate(e);
|
||||
throw propagate(e);
|
||||
}
|
||||
return result.get();
|
||||
}
|
||||
|
||||
private Iterable<String> checkNodeHasIps(NodeMetadata node) {
|
||||
Iterable<String> ips = concat(node.getPublicAddresses(), node.getPrivateAddresses());
|
||||
private FluentIterable<String> checkNodeHasIps(NodeMetadata node) {
|
||||
FluentIterable<String> ips = FluentIterable.from(concat(node.getPublicAddresses(), node.getPrivateAddresses()));
|
||||
checkState(size(ips) > 0, "node does not have IP addresses configured: " + node);
|
||||
return ips;
|
||||
}
|
||||
|
||||
private Set<HostAndPort> toHostAndPorts(Iterable<String> hosts, final int port) {
|
||||
return ImmutableSet.copyOf(transform(hosts,
|
||||
new Function<String, HostAndPort>() {
|
||||
|
||||
@Override
|
||||
public HostAndPort apply(String from) {
|
||||
return HostAndPort.fromParts(from, port);
|
||||
}
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,28 @@
|
|||
/**
|
||||
* Licensed to jclouds, Inc. (jclouds) under one or more
|
||||
* contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. jclouds 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.jclouds.compute.util;
|
||||
|
||||
import static com.google.common.base.Predicates.alwaysFalse;
|
||||
import static com.google.common.base.Predicates.alwaysTrue;
|
||||
import static com.google.common.base.Throwables.propagate;
|
||||
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
|
||||
import static java.util.concurrent.TimeUnit.MILLISECONDS;
|
||||
import static org.easymock.EasyMock.createMock;
|
||||
import static org.easymock.EasyMock.expect;
|
||||
import static org.easymock.EasyMock.replay;
|
||||
|
|
@ -13,12 +36,11 @@ import java.util.NoSuchElementException;
|
|||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.easymock.EasyMock;
|
||||
import org.jclouds.compute.domain.NodeMetadata;
|
||||
import org.jclouds.concurrent.MoreExecutors;
|
||||
import org.jclouds.compute.domain.NodeMetadataBuilder;
|
||||
import org.jclouds.predicates.SocketOpen;
|
||||
import org.testng.annotations.AfterMethod;
|
||||
import org.testng.annotations.BeforeMethod;
|
||||
|
|
@ -26,42 +48,35 @@ import org.testng.annotations.Test;
|
|||
|
||||
import com.google.common.base.Predicate;
|
||||
import com.google.common.base.Stopwatch;
|
||||
import com.google.common.base.Throwables;
|
||||
import com.google.common.collect.ImmutableMap;
|
||||
import com.google.common.collect.ImmutableSet;
|
||||
import com.google.common.net.HostAndPort;
|
||||
|
||||
@Test(singleThreaded=true)
|
||||
@Test(singleThreaded = true)
|
||||
public class ConcurrentOpenSocketFinderTest {
|
||||
|
||||
private static final long SLOW_GRACE = 500;
|
||||
private static final long EARLY_GRACE = 10;
|
||||
|
||||
private NodeMetadata node;
|
||||
private final NodeMetadata node = new NodeMetadataBuilder().id("myid").status(NodeMetadata.Status.RUNNING)
|
||||
.publicAddresses(ImmutableSet.of("1.2.3.4")).privateAddresses(ImmutableSet.of("1.2.3.5")).build();
|
||||
|
||||
private final Predicate<AtomicReference<NodeMetadata>> alwaysTrue = alwaysTrue();
|
||||
private final Predicate<AtomicReference<NodeMetadata>> alwaysFalse = alwaysFalse();
|
||||
|
||||
private SocketOpen socketTester;
|
||||
private Predicate<AtomicReference<NodeMetadata>> nodeRunning;
|
||||
private ExecutorService threadPool;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
|
||||
@BeforeMethod
|
||||
public void setUp() {
|
||||
node = createMock(NodeMetadata.class);
|
||||
expect(node.getPublicAddresses()).andReturn(ImmutableSet.of("1.2.3.4")).atLeastOnce();
|
||||
expect(node.getPrivateAddresses()).andReturn(ImmutableSet.of("1.2.3.5")).atLeastOnce();
|
||||
expect(node.getId()).andReturn("myid").anyTimes();
|
||||
|
||||
socketTester = createMock(SocketOpen.class);
|
||||
|
||||
nodeRunning = createMock(Predicate.class);
|
||||
|
||||
replay(node);
|
||||
|
||||
threadPool = Executors.newCachedThreadPool();
|
||||
}
|
||||
|
||||
@AfterMethod(alwaysRun=true)
|
||||
@AfterMethod(alwaysRun = true)
|
||||
public void tearDown() {
|
||||
if (threadPool != null) threadPool.shutdownNow();
|
||||
if (threadPool != null)
|
||||
threadPool.shutdownNow();
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -70,116 +85,106 @@ public class ConcurrentOpenSocketFinderTest {
|
|||
|
||||
expect(socketTester.apply(HostAndPort.fromParts("1.2.3.4", 22))).andReturn(false).times(2, Integer.MAX_VALUE);
|
||||
expect(socketTester.apply(HostAndPort.fromParts("1.2.3.5", 22))).andReturn(false).times(2, Integer.MAX_VALUE);
|
||||
expect(nodeRunning.apply(EasyMock.<AtomicReference<NodeMetadata>>anyObject())).andReturn(true);
|
||||
replay(socketTester);
|
||||
replay(nodeRunning);
|
||||
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, null, MoreExecutors.sameThreadExecutor());
|
||||
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, alwaysTrue, threadPool);
|
||||
|
||||
Stopwatch stopwatch = new Stopwatch();
|
||||
stopwatch.start();
|
||||
try {
|
||||
finder.findOpenSocketOnNode(node, 22, timeoutMs, TimeUnit.MILLISECONDS);
|
||||
finder.findOpenSocketOnNode(node, 22, timeoutMs, MILLISECONDS);
|
||||
fail();
|
||||
} catch (NoSuchElementException success) {
|
||||
// expected
|
||||
}
|
||||
long timetaken = stopwatch.elapsedMillis();
|
||||
|
||||
assertTrue(timetaken >= timeoutMs-EARLY_GRACE && timetaken <= timeoutMs+SLOW_GRACE, "timetaken="+timetaken);
|
||||
|
||||
verify(node);
|
||||
|
||||
assertTrue(timetaken >= timeoutMs - EARLY_GRACE && timetaken <= timeoutMs + SLOW_GRACE, "timetaken=" + timetaken);
|
||||
|
||||
verify(socketTester);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testReturnsReachable() throws Exception {
|
||||
expect(socketTester.apply(HostAndPort.fromParts("1.2.3.4", 22))).andReturn(false).once();
|
||||
expect(socketTester.apply(HostAndPort.fromParts("1.2.3.5", 22))).andReturn(true).once();
|
||||
expect(nodeRunning.apply(EasyMock.<AtomicReference<NodeMetadata>>anyObject())).andReturn(true);
|
||||
replay(socketTester);
|
||||
replay(nodeRunning);
|
||||
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, null, MoreExecutors.sameThreadExecutor());
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, alwaysTrue, threadPool);
|
||||
|
||||
HostAndPort result = finder.findOpenSocketOnNode(node, 22, 2000, TimeUnit.MILLISECONDS);
|
||||
HostAndPort result = finder.findOpenSocketOnNode(node, 22, 2000, MILLISECONDS);
|
||||
assertEquals(result, HostAndPort.fromParts("1.2.3.5", 22));
|
||||
|
||||
verify(node);
|
||||
verify(socketTester);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testChecksSocketsConcurrently() throws Exception {
|
||||
expect(nodeRunning.apply(EasyMock.<AtomicReference<NodeMetadata>>anyObject())).andReturn(true);
|
||||
replay(nodeRunning);
|
||||
|
||||
// Can't use mock+answer for concurrency tests; EasyMock uses lock in ReplayState
|
||||
// Can't use mock+answer for concurrency tests; EasyMock uses lock in
|
||||
// ReplayState
|
||||
ControllableSocketOpen socketTester = new ControllableSocketOpen(ImmutableMap.of(
|
||||
HostAndPort.fromParts("1.2.3.4", 22), new SlowCallable<Boolean>(true, 1500),
|
||||
HostAndPort.fromParts("1.2.3.5", 22), new SlowCallable<Boolean>(true, 1000)));
|
||||
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, null, threadPool);
|
||||
HostAndPort.fromParts("1.2.3.4", 22), new SlowCallable<Boolean>(true, 1500),
|
||||
HostAndPort.fromParts("1.2.3.5", 22), new SlowCallable<Boolean>(true, 1000)));
|
||||
|
||||
HostAndPort result = finder.findOpenSocketOnNode(node, 22, 2000, TimeUnit.MILLISECONDS);
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, alwaysTrue, threadPool);
|
||||
|
||||
HostAndPort result = finder.findOpenSocketOnNode(node, 22, 2000, MILLISECONDS);
|
||||
assertEquals(result, HostAndPort.fromParts("1.2.3.5", 22));
|
||||
verify(node);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testAbortsWhenNodeNotRunning() throws Exception {
|
||||
expect(socketTester.apply(EasyMock.<HostAndPort>anyObject())).andReturn(false);
|
||||
expect(nodeRunning.apply(EasyMock.<AtomicReference<NodeMetadata>>anyObject())).andReturn(false);
|
||||
expect(socketTester.apply(EasyMock.<HostAndPort> anyObject())).andReturn(false);
|
||||
replay(socketTester);
|
||||
replay(nodeRunning);
|
||||
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, nodeRunning, MoreExecutors.sameThreadExecutor());
|
||||
|
||||
|
||||
OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketTester, alwaysFalse, threadPool);
|
||||
|
||||
Stopwatch stopwatch = new Stopwatch();
|
||||
stopwatch.start();
|
||||
try {
|
||||
finder.findOpenSocketOnNode(node, 22, 2000, TimeUnit.MILLISECONDS);
|
||||
finder.findOpenSocketOnNode(node, 22, 2000, MILLISECONDS);
|
||||
fail();
|
||||
} catch (NoSuchElementException e) {
|
||||
// success
|
||||
// Note: don't get the "no longer running" message, because logged+swallowed by RetryablePredicate
|
||||
// Note: don't get the "no longer running" message, because
|
||||
// logged+swallowed by RetryablePredicate
|
||||
}
|
||||
long timetaken = stopwatch.elapsedMillis();
|
||||
|
||||
assertTrue(timetaken <= SLOW_GRACE, "timetaken="+timetaken);
|
||||
|
||||
verify(node);
|
||||
assertTrue(timetaken <= SLOW_GRACE, "timetaken=" + timetaken);
|
||||
|
||||
verify(socketTester);
|
||||
verify(nodeRunning);
|
||||
}
|
||||
|
||||
private static class SlowCallable<T> implements Callable<T> {
|
||||
private final T result;
|
||||
private final long delay;
|
||||
|
||||
|
||||
SlowCallable(T result, long delay) {
|
||||
this.result = result;
|
||||
this.delay = delay;
|
||||
}
|
||||
|
||||
@Override
|
||||
public T call() throws Exception {
|
||||
Thread.sleep(delay);
|
||||
sleepUninterruptibly(delay, MILLISECONDS);
|
||||
return result;
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
private static class ControllableSocketOpen implements SocketOpen {
|
||||
private final Map<HostAndPort, ? extends Callable<Boolean>> answers;
|
||||
|
||||
|
||||
ControllableSocketOpen(Map<HostAndPort, ? extends Callable<Boolean>> answers) {
|
||||
this.answers = answers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean apply(HostAndPort input) {
|
||||
try {
|
||||
return answers.get(input).call();
|
||||
} catch (Exception e) {
|
||||
throw Throwables.propagate(e);
|
||||
throw propagate(e);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
|
|
|||
Loading…
Reference in New Issue