Author: frm
Date: Tue Oct 4 12:13:54 2016
New Revision: 1763269
URL: http://svn.apache.org/viewvc?rev=1763269&view=rev
Log:
OAK-4884 - Make bind/connect/close synchronous for the standby client, server
and proxy
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/NetworkErrorProxy.java
- copied, changed from r1763254,
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NetworkErrorProxy.java
Removed:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NetworkErrorProxy.java
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyServer.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/BrokenNetworkTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/DataStoreTestBase.java
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java?rev=1763269&r1=1763268&r2=1763269&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
Tue Oct 4 12:13:54 2016
@@ -47,9 +47,13 @@ import org.apache.jackrabbit.oak.segment
import
org.apache.jackrabbit.oak.segment.standby.codec.GetSegmentRequestEncoder;
import org.apache.jackrabbit.oak.segment.standby.codec.GetSegmentResponse;
import org.apache.jackrabbit.oak.segment.standby.codec.ResponseDecoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
class StandbyClient implements AutoCloseable {
+ private static final Logger log =
LoggerFactory.getLogger(StandbyClient.class);
+
private final BlockingQueue<GetHeadResponse> headQueue = new
LinkedBlockingDeque<>();
private final BlockingQueue<GetSegmentResponse> segmentQueue = new
LinkedBlockingDeque<>();
@@ -135,10 +139,20 @@ class StandbyClient implements AutoClose
@Override
public void close() throws InterruptedException {
- channel.close().sync();
+ if (channel.close().awaitUninterruptibly(1, TimeUnit.SECONDS)) {
+ log.debug("Channel closed");
+ } else {
+ log.debug("Channel close timed out");
+ }
+
channel = null;
- group.shutdownGracefully();
+ if (group.shutdownGracefully().awaitUninterruptibly(1,
TimeUnit.SECONDS)) {
+ log.debug("Group shut down");
+ } else {
+ log.debug("Group shutdown timed out");
+ }
+
group = null;
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyServer.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyServer.java?rev=1763269&r1=1763268&r2=1763269&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyServer.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyServer.java
Tue Oct 4 12:13:54 2016
@@ -176,7 +176,7 @@ class StandbyServer implements AutoClose
public void start() {
channelFuture = b.bind(port);
- if (channelFuture.awaitUninterruptibly(10, TimeUnit.SECONDS)) {
+ if (channelFuture.awaitUninterruptibly(1, TimeUnit.SECONDS)) {
onTimelyConnect();
} else {
onConnectTimeOut();
@@ -184,21 +184,34 @@ class StandbyServer implements AutoClose
}
public void stop() {
- channelFuture.channel().disconnect();
+ if (channelFuture.channel().disconnect().awaitUninterruptibly(1,
TimeUnit.SECONDS)) {
+ log.debug("Channel disconnected");
+ } else {
+ log.debug("Channel disconnect timed out");
+ }
}
@Override
public void close() {
stop();
- if (bossGroup != null && !bossGroup.isShuttingDown()) {
- bossGroup.shutdownGracefully(0, 1,
TimeUnit.SECONDS).syncUninterruptibly();
+ if (shutDown(bossGroup)) {
+ log.debug("Boss group shut down");
+ } else {
+ log.debug("Boss group shutdown timed out");
}
- if (workerGroup != null && !workerGroup.isShuttingDown()) {
- workerGroup.shutdownGracefully(0, 1,
TimeUnit.SECONDS).syncUninterruptibly();
+
+ if (shutDown(workerGroup)) {
+ log.debug("Worker group shut down");
+ } else {
+ log.debug("Worker group shutdown timed out");
}
}
+ private boolean shutDown(EventLoopGroup group) {
+ return group.shutdownGracefully(0, 1,
TimeUnit.SECONDS).awaitUninterruptibly(1, TimeUnit.SECONDS);
+ }
+
private void onTimelyConnect() {
if (channelFuture.isSuccess()) {
log.debug("Binding was successful");
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/BrokenNetworkTest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/BrokenNetworkTest.java?rev=1763269&r1=1763268&r2=1763269&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/BrokenNetworkTest.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/BrokenNetworkTest.java
Tue Oct 4 12:13:54 2016
@@ -23,7 +23,6 @@ import static org.apache.jackrabbit.oak.
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import org.apache.jackrabbit.oak.segment.NetworkErrorProxy;
import org.apache.jackrabbit.oak.segment.SegmentNodeStoreBuilders;
import org.apache.jackrabbit.oak.segment.standby.client.StandbyClientSync;
import org.apache.jackrabbit.oak.segment.standby.server.StandbyServerSync;
@@ -115,21 +114,23 @@ public class BrokenNetworkTest extends T
}
private void useProxy(boolean ssl, int skipPosition, int skipBytes, int
flipPosition, boolean intermediateChange) throws Exception {
- NetworkErrorProxy p = new NetworkErrorProxy(proxyPort, LOCALHOST,
port);
- p.skipBytes(skipPosition, skipBytes);
- p.flipByte(flipPosition);
- p.run();
-
NodeStore store = SegmentNodeStoreBuilders.builder(storeS).build();
- final StandbyServerSync serverSync = new StandbyServerSync(port,
storeS, ssl);
- serverSync.start();
addTestContent(store, "server");
storeS.flush(); // this speeds up the test a little bit...
- StandbyClientSync clientSync = newStandbyClientSync(storeC, proxyPort,
ssl);
- clientSync.run();
+ try (
+ NetworkErrorProxy p = new NetworkErrorProxy(proxyPort,
LOCALHOST, port);
+ StandbyServerSync serverSync = new StandbyServerSync(port,
storeS, ssl);
+ StandbyClientSync clientSync = newStandbyClientSync(storeC,
proxyPort, ssl);
+ ) {
+ p.skipBytes(skipPosition, skipBytes);
+ p.flipByte(flipPosition);
+ p.run();
+
+ serverSync.start();
+
+ clientSync.run();
- try {
if (skipBytes > 0 || flipPosition >= 0) {
assertFalse("stores are not expected to be equal",
storeS.getHead().equals(storeC.getHead()));
assertEquals(storeC2.getHead(), storeC.getHead());
@@ -141,11 +142,9 @@ public class BrokenNetworkTest extends T
}
clientSync.run();
}
+
assertEquals(storeS.getHead(), storeC.getHead());
- } finally {
- serverSync.close();
- clientSync.close();
- p.close();
}
}
+
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/DataStoreTestBase.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/DataStoreTestBase.java?rev=1763269&r1=1763268&r2=1763269&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/DataStoreTestBase.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/DataStoreTestBase.java
Tue Oct 4 12:13:54 2016
@@ -38,7 +38,6 @@ import org.apache.jackrabbit.oak.api.Com
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.api.Type;
import org.apache.jackrabbit.oak.plugins.blob.datastore.DataStoreBlobStore;
-import org.apache.jackrabbit.oak.segment.NetworkErrorProxy;
import org.apache.jackrabbit.oak.segment.SegmentNodeStoreBuilders;
import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.segment.file.FileStoreBuilder;
Copied:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/NetworkErrorProxy.java
(from r1763254,
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NetworkErrorProxy.java)
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/NetworkErrorProxy.java?p2=jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/NetworkErrorProxy.java&p1=jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NetworkErrorProxy.java&r1=1763254&r2=1763269&rev=1763269&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NetworkErrorProxy.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/NetworkErrorProxy.java
Tue Oct 4 12:13:54 2016
@@ -17,8 +17,9 @@
* under the License.
*/
-package org.apache.jackrabbit.oak.segment;
+package org.apache.jackrabbit.oak.segment.standby;
+import java.io.Closeable;
import java.util.concurrent.TimeUnit;
import io.netty.bootstrap.Bootstrap;
@@ -37,10 +38,9 @@ import io.netty.channel.socket.nio.NioSo
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class NetworkErrorProxy {
+public class NetworkErrorProxy implements Closeable {
- static final Logger log = LoggerFactory
- .getLogger(NetworkErrorProxy.class);
+ static final Logger log = LoggerFactory.getLogger(NetworkErrorProxy.class);
private final int inboundPort;
@@ -52,9 +52,9 @@ public class NetworkErrorProxy {
private ForwardHandler fh;
- EventLoopGroup bossGroup = new NioEventLoopGroup();
+ private final EventLoopGroup bossGroup = new NioEventLoopGroup();
- EventLoopGroup workerGroup = new NioEventLoopGroup();
+ private final EventLoopGroup workerGroup = new NioEventLoopGroup();
public NetworkErrorProxy(int inboundPort, String outboundHost, int
outboundPort) {
this.inboundPort = inboundPort;
@@ -74,22 +74,21 @@ public class NetworkErrorProxy {
public void run() throws Exception {
try {
- ServerBootstrap b = new ServerBootstrap();
- b.group(bossGroup, workerGroup)
+ ServerBootstrap b = new ServerBootstrap()
+ .group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws
Exception {
- ch.pipeline().addLast(NetworkErrorProxy.this.fh);
+ ch.pipeline().addLast(fh);
}
+
});
f = b.bind(this.inboundPort).sync();
} catch (Exception e) {
- log.warn(
- "Unable to start proxy on port " + inboundPort + ": "
- + e.getMessage(), e);
+ log.warn(String.format("Unable to start proxy on port %d",
inboundPort), e);
}
}
@@ -97,20 +96,38 @@ public class NetworkErrorProxy {
if (f == null) {
throw new Exception("proxy not started");
}
- f.channel().disconnect();
- this.fh = new ForwardHandler(NetworkErrorProxy.this.host,
NetworkErrorProxy.this.outboundPort);
+
+ if (f.channel().disconnect().awaitUninterruptibly(10,
TimeUnit.SECONDS)) {
+ log.debug("Channel disconnected");
+ } else {
+ log.debug("Channel disconnect timed out");
+ }
+
+ fh = new ForwardHandler(NetworkErrorProxy.this.host,
NetworkErrorProxy.this.outboundPort);
+
run();
}
+ @Override
public void close() {
if (f != null) {
- f.channel().close().syncUninterruptibly();
+ if (f.channel().close().awaitUninterruptibly(10,
TimeUnit.SECONDS)) {
+ log.debug("Channel closed");
+ } else {
+ log.debug("Channel close timed out");
+ }
}
- if (bossGroup != null && !bossGroup.isShuttingDown()) {
- bossGroup.shutdownGracefully(0, 150,
TimeUnit.MILLISECONDS).syncUninterruptibly();
+
+ if (bossGroup.shutdownGracefully(0, 150,
TimeUnit.MILLISECONDS).awaitUninterruptibly(1, TimeUnit.SECONDS)) {
+ log.debug("Boss group shut down");
+ } else {
+ log.debug("Boss group shutdown timed out");
}
- if (workerGroup != null && !workerGroup.isShuttingDown()) {
- workerGroup.shutdownGracefully(0, 150,
TimeUnit.MILLISECONDS).syncUninterruptibly();
+
+ if (workerGroup.shutdownGracefully(0, 150,
TimeUnit.MILLISECONDS).awaitUninterruptibly(1, TimeUnit.SECONDS)) {
+ log.debug("Worker group shut down");
+ } else {
+ log.debug("Worker group shutdown timed out");
}
}
}
