Skip to content

Commit

Permalink
(issue #29) increased logging, reduced dataset size for busy test
Browse files Browse the repository at this point in the history
  • Loading branch information
@bluestreak01
bluestreak01 committed Jan 22, 2015
1 parent c330e1e commit b1a072a
Show file tree
Hide file tree
Showing 4 changed files with 18 additions and 18 deletions.
2 changes: 1 addition & 1 deletion nfsdb-core/src/main/java/com/nfsdb/net/JournalServerAgent.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -321,7 +321,7 @@ private boolean processJournalEvents(final WritableByteChannel channel, boolean
if (server.isRunning()) {
commandProducer.write(channel, Command.SERVER_HEARTBEAT);
server.getLogger().msg()
.setLevel(ServerLogMsg.Level.INFO)
.setLevel(ServerLogMsg.Level.TRACE)
.setSocketAddress(socketAddress)
.setMessage("Heartbeat")
.send();
Expand Down
10 changes: 10 additions & 0 deletions nfsdb-core/src/main/java/com/nfsdb/net/cluster/ClusterController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -109,13 +109,15 @@ private void up() throws JournalNetworkException {

try {
if (activeNode != null) {
LOGGER.info(thisNode() + " There is active node already %s. Yielding");
setupClient(activeNode);
return;
}
} catch (JournalNetworkException ignore) {
LOGGER.info("Exception during initial server acquisition. It is safe to ignore: %s", ignore.getMessage());
}

LOGGER.info(thisNode() + " Starting server");
server = new JournalServer(new ServerConfig() {{
setHostname(thisNode().getAddress());
setEnableMulticast(false);
Expand All @@ -126,16 +128,20 @@ private void up() throws JournalNetworkException {
}
server.start();


if ((activeNode = getActiveNode()) != null && !client.voteInstance(instance)) {
LOGGER.info(thisNode() + " Lost tie-break vote, becoming a client");
// don't stop server explicitly, it wil shut down after being voted out
setupClient(activeNode);
return;
}

if (client != null) {
LOGGER.info(thisNode() + " Stopping client remnants");
client.halt();
client = null;
}
LOGGER.info(thisNode() + " Activating callback");
listener.onNodeActive();
}

Expand Down Expand Up @@ -167,11 +173,14 @@ private ClusterNode getActiveNode() {
}

private void setupClient(ClusterNode node) throws JournalNetworkException {

LOGGER.info(thisNode() + " Subscribing journals");
for (int i = 0, sz = writers.size(); i < sz; i++) {
JournalWriter w = writers.get(i);
client.subscribe(w.getKey(), w, null);
}

LOGGER.info(thisNode() + " Starting client");
client.setDisconnectCallback(new JournalClient.DisconnectCallback() {
@Override
public void onDisconnect(JournalClient.DisconnectReason reason) {
Expand All @@ -183,6 +192,7 @@ public void onDisconnect(JournalClient.DisconnectReason reason) {
}).start();

if (listener != null) {
LOGGER.info(thisNode() + " Notifying callback of standby state");
listener.onNodeStandingBy(node);
}

Expand Down
10 changes: 0 additions & 10 deletions nfsdb-core/src/main/java/com/nfsdb/net/config/ClientConfig.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -94,16 +94,6 @@ public SocketChannel openSocketChannel() throws JournalNetworkException {
.setOption(StandardSocketOptions.SO_RCVBUF, getSoRcvBuf())
.setOption(StandardSocketOptions.SO_LINGER, getLinger());

// linux doubles buffer size, which is returned
if (channel.getOption(StandardSocketOptions.SO_SNDBUF) != getSoSndBuf()
&& channel.getOption(StandardSocketOptions.SO_SNDBUF) != getSoSndBuf() * 2) {
LOGGER.warn("SO_SNDBUF value is ignored");
}
if (channel.getOption(StandardSocketOptions.SO_RCVBUF) != getSoRcvBuf()
&& channel.getOption(StandardSocketOptions.SO_RCVBUF) != getSoRcvBuf() * 2) {
LOGGER.warn("SO_RCVBUF value is ignored");
}

LOGGER.info("Connected to %s", address);
return channel;
} catch (UnresolvedAddressException e) {
Expand Down
14 changes: 7 additions & 7 deletions nfsdb-core/src/test/java/com/nfsdb/net/ClusterControllerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ public void testTiebreakFailOver() throws Exception {
active2Latch.await(10, TimeUnit.SECONDS);
Assert.assertEquals("Node 2 is expected to become active", 0, active2Latch.getCount());

active1Latch.await(200, TimeUnit.MILLISECONDS);
active1Latch.await(1, TimeUnit.SECONDS);
Assert.assertEquals("Node 1 active() callback should not have been called", 1, active1Latch.getCount());

standby1Latch.await(200, TimeUnit.MILLISECONDS);
Expand All @@ -94,7 +94,7 @@ public void testTiebreakFailOver() throws Exception {
controller2.halt();


shutdown2.await(1, TimeUnit.SECONDS);
shutdown2.await(5, TimeUnit.SECONDS);
Assert.assertEquals("Controller 2 should have shut down", 0, shutdown2.getCount());

active1Latch.await(10, TimeUnit.SECONDS);
Expand Down Expand Up @@ -201,14 +201,14 @@ public void testBusyFailOver() throws Exception {
@Override
public void onNodeActive() {
try {
TestUtils.generateQuoteData(writer1, 1000000);
TestUtils.generateQuoteData(writer1, 1000000, writer1.getMaxTimestamp());
TestUtils.generateQuoteData(writer1, 100000);
TestUtils.generateQuoteData(writer1, 100000, writer1.getMaxTimestamp());
writer1.commit();
TestUtils.generateQuoteData(writer1, 1000000, writer1.getMaxTimestamp());
TestUtils.generateQuoteData(writer1, 100000, writer1.getMaxTimestamp());
writer1.commit();
TestUtils.generateQuoteData(writer1, 1000000, writer1.getMaxTimestamp());
TestUtils.generateQuoteData(writer1, 100000, writer1.getMaxTimestamp());
writer1.commit();
TestUtils.generateQuoteData(writer1, 1000000, writer1.getMaxTimestamp());
TestUtils.generateQuoteData(writer1, 100000, writer1.getMaxTimestamp());
writer1.commit();
expected.set(writer1.size());
active1.countDown();
Expand Down

0 comments on commit b1a072a

Please sign in to comment.