Menu
▾
▴
[Bigdata-commit] SF.net SVN: bigdata:[7601] branches/BIGDATA_RELEASE_1_3_0
|
From: <tho...@us...> - 2013-12-03 17:58:47
|
Revision: 7601
http://bigdata.svn.sourceforge.net/bigdata/?rev=7601&view=rev
Author: thompsonbry
Date: 2013-12-03 17:58:35 +0000 (Tue, 03 Dec 2013)
Log Message:
-----------
Reconciling changes with Martyn for the postHACommit(), meta-bits maintenance on replication and leader failure, and in the HA test suite.
Changes to:
- StoreState interface
- IHABufferStrategy.getStoreState()
- RWStrategy.getStoreState()
- WormStrategy.getStoreState()
- FixedAllocator log @ WARN
- RWStore (for unpacking more information from the new root block in postHACommit()).
- HAJournalTest (for getStoreState() method on HAGlueTest).
- AbstractHA3JournalServerTestCase.assertStoreStates()
- TestHA3ChangeLeader (but disabled the kill-based test since this can trigger the socket resync problem and that code was not brought across)
- BytesUtil.toHexString() allows a null argument.
See #778.
Modified Paths:
--------------
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/btree/BytesUtil.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractBufferStrategy.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/IHABufferStrategy.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/RWStrategy.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/WORMStrategy.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/FixedAllocator.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/RWStore.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3ChangeLeader.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/btree/BytesUtil.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/btree/BytesUtil.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/btree/BytesUtil.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -1576,6 +1576,9 @@
*/
static public String toHexString(final byte[] buf) {
+ if (buf == null)
+ return "NULL";
+
return toHexString(buf, buf.length);
}
@@ -1591,6 +1594,10 @@
* @return The hex string.
*/
static public String toHexString(final byte[] buf, int n) {
+
+ if (buf == null)
+ return "NULL";
+
n = n < buf.length ? n : buf.length;
final StringBuffer out = new StringBuffer();
for (int i = 0; i < n; i++) {
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractBufferStrategy.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractBufferStrategy.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractBufferStrategy.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -23,11 +23,7 @@
*/
package com.bigdata.journal;
-import java.io.ByteArrayOutputStream;
-import java.io.DataOutputStream;
-import java.io.EOFException;
import java.io.IOException;
-import java.io.InputStream;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
@@ -42,8 +38,6 @@
import com.bigdata.mdi.IResourceMetadata;
import com.bigdata.rawstore.AbstractRawWormStore;
import com.bigdata.rawstore.Bytes;
-import com.bigdata.rawstore.IAllocationContext;
-import com.bigdata.rawstore.IPSOutputStream;
import com.bigdata.rawstore.IRawStore;
import com.bigdata.rawstore.WormAddressManager;
import com.bigdata.resources.ResourceManager;
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/IHABufferStrategy.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/IHABufferStrategy.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/IHABufferStrategy.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -280,4 +280,11 @@
*/
WriteCacheService getWriteCacheService();
+ /**
+ * A StoreState object references critical transient data that can be used
+ * to determine a degree of consistency between stores, specifically for an
+ * HA context.
+ */
+ StoreState getStoreState();
+
}
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/RWStrategy.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/RWStrategy.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/RWStrategy.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -906,6 +906,11 @@
return m_store.getWriteCacheService();
}
+ @Override
+ public StoreState getStoreState() {
+ return m_store.getStoreState();
+ }
+
// @Override
// public boolean isFlushed() {
// return m_store.isFlushed();
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/WORMStrategy.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/WORMStrategy.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/WORMStrategy.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -2942,6 +2942,26 @@
// m_rebuildSequence++;
}
+ @Override
+ public StoreState getStoreState() {
+ return new WormStoreState();
+ }
+
+ public static class WormStoreState implements StoreState {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (obj == null || !(obj instanceof WormStoreState))
+ return false;
+ final WormStoreState other = (WormStoreState) obj;
+ // Nothing to compare.
+ return true;
+ }
+
+ }
+
// @Override
// public void prepareForRebuild(HARebuildRequest req) {
// assert m_rebuildRequest == null;
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/FixedAllocator.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/FixedAllocator.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/FixedAllocator.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -180,6 +180,8 @@
m_store.showWriteCacheDebug(paddr);
+ log.warn("Physical address " + paddr + " not accessible for Allocator of size " + m_size);
+
return 0L;
}
}
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/RWStore.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/RWStore.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/rwstore/RWStore.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -100,6 +100,7 @@
import com.bigdata.journal.IHABufferStrategy;
import com.bigdata.journal.IRootBlockView;
import com.bigdata.journal.RootBlockView;
+import com.bigdata.journal.StoreState;
import com.bigdata.journal.StoreTypeEnum;
import com.bigdata.quorum.Quorum;
import com.bigdata.quorum.QuorumException;
@@ -1219,31 +1220,44 @@
}
- /*
- * Utility to encapsulate RootBlock interpreation
+ /**
+ * Utility to encapsulate RootBlock interpretation.
*/
- static class RootBlockInfo {
+ static private class RootBlockInfo {
- static int nextAllocation(final IRootBlockView rb) {
- final long nxtOffset = rb.getNextOffset();
-
- // next allocation to be made (in -32K units).
- final int ret = -(int) (nxtOffset >> 32);
-
- /*
- * Skip the first 32K in the file. The root blocks live here but
- * nothing else.
- */
- return ret == 0 ? -(1 + META_ALLOCATION) : ret;
- }
+// int nextAllocation(final IRootBlockView rb) {
+// final long nxtOffset = rb.getNextOffset();
+//
+// // next allocation to be made (in -32K units).
+// final int ret = -(int) (nxtOffset >> 32);
+//
+// /*
+// * Skip the first 32K in the file. The root blocks live here but
+// * nothing else.
+// */
+// return ret == 0 ? -(1 + META_ALLOCATION) : ret;
+// }
- /*
+ /**
+ * Used to transparently re-open the backing channel if it has been closed
+ * by an interrupt during an IO.
+ */
+ private final ReopenFileChannel m_reopener;
+ /**
* Meta-Allocations stored as {int address; int[8] bits}, so each block
* holds 8*32=256 allocation slots of 1K totaling 256K.
- *
+ * <p>
* The returned int array is a flattened list of these int[9] blocks
*/
- static int[] metabits(final IRootBlockView rb, final ReopenFileChannel reopener) throws IOException {
+ private final int[] m_metabits;
+ private final long m_storageStatsAddr;
+ private final long m_lastDeferredReleaseTime;
+
+ RootBlockInfo(final IRootBlockView rb,
+ final ReopenFileChannel reopener) throws IOException {
+
+ this.m_reopener = reopener;
+
final long rawmbaddr = rb.getMetaBitsAddr();
/*
@@ -1265,17 +1279,17 @@
*/
final byte[] buf = new byte[metaBitsStore * 4];
- FileChannelUtility.readAll(reopener, ByteBuffer.wrap(buf), pmaddr);
+ FileChannelUtility.readAll(m_reopener, ByteBuffer.wrap(buf), pmaddr);
final DataInputStream strBuf = new DataInputStream(new ByteArrayInputStream(buf));
// Can handle minor store version incompatibility
strBuf.readInt(); // STORE VERSION
- strBuf.readLong(); // Last Deferred Release Time
+ m_lastDeferredReleaseTime = strBuf.readLong(); // Last Deferred Release Time
strBuf.readInt(); // cDefaultMetaBitsSize
final int allocBlocks = strBuf.readInt();
- strBuf.readLong(); // m_storageStatsAddr
+ m_storageStatsAddr = strBuf.readLong(); // m_storageStatsAddr
// step over those reserved ints
for (int i = 0; i < cReservedMetaBits; i++) {
@@ -1291,7 +1305,7 @@
// Must be multiple of 9
assert metaBitsSize % 9 == 0;
- int[] ret = new int[metaBitsSize];
+ final int[] ret = new int[metaBitsSize];
for (int i = 0; i < metaBitsSize; i++) {
ret[i] = strBuf.readInt();
}
@@ -1300,8 +1314,9 @@
* Meta-Allocations stored as {int address; int[8] bits}, so each block
* holds 8*32=256 allocation slots of 1K totaling 256K.
*/
- return ret;
+ m_metabits = ret;
}
+
}
/**
@@ -3157,6 +3172,13 @@
log.trace("commitChanges for: " + m_nextAllocation + ", "
+ m_metaBitsAddr + ", active contexts: "
+ m_contexts.size());
+
+ if (log.isDebugEnabled() && m_quorum.isHighlyAvailable()) {
+
+ log.debug(showAllocatorList());
+
+ }
+
}
/**
@@ -6216,14 +6238,40 @@
log.trace("Allocator " + index + ", size: " + xfa.m_size + ", startAddress: " + xfa.getStartAddr() + ", allocated: " + (xfa.getAllocatedSlots()/xfa.m_size));
}
}
-
+
+ // Update m_metaBits addr and m_nextAllocation to ensure able to allocate as well as read!
+ {
+ final long nxtOffset = rbv.getNextOffset();
+
+ // next allocation to be made (in -32K units).
+ m_nextAllocation = -(int) (nxtOffset >> 32);
+
+ if (m_nextAllocation == 0) {
+ throw new IllegalStateException("Invalid state for non-empty store");
+ }
+
+ m_committedNextAllocation = m_nextAllocation;
+
+ final long savedMetaBitsAddr = m_metaBitsAddr;
+ // latched offset of the metabits region.
+ m_metaBitsAddr = -(int) nxtOffset;
+
+ if (savedMetaBitsAddr != m_metaBitsAddr)
+ log.warn("Old metaBitsAddr: " + savedMetaBitsAddr + ", new metaBitsAddr: " + m_metaBitsAddr);
+ }
+
final ArrayList<FixedAllocator> nallocs = new ArrayList<FixedAllocator>();
// current metabits
final int[] oldmetabits = m_metaBits;
// new metabits
- m_metaBits = RootBlockInfo.metabits(rbv, m_reopener);
+ final RootBlockInfo rbi = new RootBlockInfo(rbv, m_reopener);
+ m_metaBits = rbi.m_metabits;
+ // and grab the last deferred release and storageStats!
+ m_lastDeferredReleaseTime = rbi.m_lastDeferredReleaseTime;
+ m_storageStatsAddr = rbi.m_storageStatsAddr;
+
if(log.isTraceEnabled())
log.trace("Metabits length: " + m_metaBits.length);
@@ -6925,6 +6973,16 @@
}
+ private String showAllocatorList() {
+ final StringBuilder sb = new StringBuilder();
+
+ for (int index = 0; index < m_allocs.size(); index++) {
+ final FixedAllocator xfa = m_allocs.get(index);
+ sb.append("Allocator " + index + ", size: " + xfa.m_size + ", startAddress: " + xfa.getStartAddr() + ", allocated: " + xfa.getAllocatedSlots() + "\n");
+ }
+
+ return sb.toString();
+ }
// /**
// *
// * @return whether WCS is flushed
@@ -6935,6 +6993,79 @@
// return this.m_writeCacheService.isFlushed();
// }
+ public static class RWStoreState implements StoreState {
+
+ /**
+ * Generated ID
+ */
+ private static final long serialVersionUID = 4315400143557397323L;
+
+ /*
+ * Transient state necessary for consistent ha leader transition
+ */
+ private final int m_fileSize;
+ private final int m_nextAllocation;
+ private final int m_committedNextAllocation;
+ private final long m_minReleaseAge;
+ private final long m_lastDeferredReleaseTime;
+ private final long m_storageStatsAddr;
+ private final int m_allocsSize;
+ private final int m_metaBitsAddr;
+ private final int m_metaBitsSize;
+
+ private RWStoreState(final RWStore store) {
+ m_fileSize = store.m_fileSize;
+ m_nextAllocation = store.m_nextAllocation;
+ m_committedNextAllocation = store.m_committedNextAllocation;
+ m_minReleaseAge = store.m_minReleaseAge;
+ m_lastDeferredReleaseTime = store.m_lastDeferredReleaseTime;
+ m_storageStatsAddr = store.m_storageStatsAddr;
+ m_allocsSize = store.m_allocs.size();
+ m_metaBitsAddr = store.m_metaBitsAddr;
+ m_metaBitsSize = store.m_metaBits.length;
+ }
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (obj == null || !(obj instanceof RWStoreState))
+ return false;
+ final RWStoreState other = (RWStoreState) obj;
+ return m_fileSize == other.m_fileSize
+ && m_nextAllocation == other.m_nextAllocation
+ && m_committedNextAllocation == other.m_committedNextAllocation
+ && m_minReleaseAge == other.m_minReleaseAge
+ && m_lastDeferredReleaseTime == other.m_lastDeferredReleaseTime
+ && m_storageStatsAddr == other.m_storageStatsAddr
+ && m_allocsSize == other.m_allocsSize
+ && m_metaBitsAddr == other.m_metaBitsAddr
+ && m_metaBitsSize == other.m_metaBitsSize;
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder();
+
+ sb.append("RWStoreState\n");
+ sb.append("fileSize: " + m_fileSize + "\n");
+ sb.append("nextAllocation: " + m_nextAllocation + "\n");
+ sb.append("committedNextAllocation: " + m_committedNextAllocation + "\n");
+ sb.append("minReleaseAge: " + m_minReleaseAge + "\n");
+ sb.append("lastDeferredReleaseTime: " + m_lastDeferredReleaseTime + "\n");
+ sb.append("storageStatsAddr: " + m_storageStatsAddr + "\n");
+ sb.append("allocsSize: " + m_allocsSize + "\n");
+ sb.append("metaBitsAddr: " + m_metaBitsAddr + "\n");
+ sb.append("metaBitsSize: " + m_metaBitsSize + "\n");
+
+ return sb.toString();
+ }
+ }
+
+ public StoreState getStoreState() {
+ final RWStoreState ret = new RWStoreState(this);
+
+ return ret;
+ }
+
// public void prepareForRebuild(final HARebuildRequest req) {
// assert m_rebuildRequest == null;
//
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -85,6 +85,7 @@
import com.bigdata.jini.util.ConfigMath;
import com.bigdata.jini.util.JiniUtil;
import com.bigdata.journal.IRootBlockView;
+import com.bigdata.journal.StoreState;
import com.bigdata.journal.jini.ha.HAJournalServer.ConfigurationOptions;
import com.bigdata.journal.jini.ha.HAJournalTest.HAGlueTest;
import com.bigdata.quorum.AbstractQuorumClient;
@@ -2688,6 +2689,39 @@
}
+ protected void assertStoreStates(final HAGlue[] services) throws IOException {
+ if (services.length < 2)
+ return; // nothing to compare
+
+ final StoreState test = ((HAGlueTest) services[0]).getStoreState();
+ final String tname = serviceName(services[0]);
+
+ for (int s = 1; s < services.length; s++) {
+ final StoreState other = ((HAGlueTest) services[s]).getStoreState();
+
+ if (!test.equals(other)) {
+ final String oname = serviceName(services[s]);
+ final String msg = "StoreState mismatch \n" + tname + "\n"
+ + test.toString() + "\n" + oname + "\n"
+ + other.toString();
+ fail(msg);
+ }
+ }
+ }
+
+ protected String serviceName(final HAGlue s) {
+ if (s == serverA) {
+ return "serverA";
+ } else if (s == serverB) {
+ return "serverB";
+ } else if (s == serverC) {
+ return "serverC";
+ } else {
+ return "NA";
+ }
+ }
+
+
/**
* Task loads a large data set.
*/
@@ -2695,6 +2729,7 @@
private final long token;
private final boolean reallyLargeLoad;
+ private final boolean dropAll;
/**
* Large load.
@@ -2708,6 +2743,10 @@
}
+
+ public LargeLoadTask(long token, boolean reallyLargeLoad) {
+ this(token, reallyLargeLoad, true/*dropAll*/);
+ }
/**
* Either large or really large load.
*
@@ -2716,17 +2755,20 @@
* @param reallyLargeLoad
* if we will also load the 3 degrees of freedom file.
*/
- public LargeLoadTask(final long token, final boolean reallyLargeLoad) {
+ public LargeLoadTask(final long token, final boolean reallyLargeLoad, final boolean dropAll) {
this.token = token;
this.reallyLargeLoad = reallyLargeLoad;
+ this.dropAll = dropAll;
+
}
public Void call() throws Exception {
final StringBuilder sb = new StringBuilder();
+ if (dropAll)
sb.append("DROP ALL;\n");
sb.append("LOAD <" + getFoafFileUrl("data-0.nq.gz") + ">;\n");
sb.append("LOAD <" + getFoafFileUrl("data-1.nq.gz") + ">;\n");
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -84,8 +84,10 @@
import com.bigdata.ha.msg.IHAWriteSetStateRequest;
import com.bigdata.ha.msg.IHAWriteSetStateResponse;
import com.bigdata.journal.AbstractJournal;
+import com.bigdata.journal.IHABufferStrategy;
import com.bigdata.journal.IRootBlockView;
import com.bigdata.journal.ITx;
+import com.bigdata.journal.StoreState;
import com.bigdata.journal.jini.ha.HAJournalServer.HAQuorumService;
import com.bigdata.journal.jini.ha.HAJournalServer.RunStateEnum;
import com.bigdata.quorum.AsynchronousQuorumCloseException;
@@ -308,6 +310,11 @@
*/
public void simpleTransaction_abort() throws IOException, Exception;
+ /**
+ * Supports consistency checking between HA services
+ */
+ public StoreState getStoreState() throws IOException;
+
}
/**
@@ -1164,6 +1171,12 @@
}
+ @Override
+ public StoreState getStoreState() throws IOException {
+ return ((IHABufferStrategy) (getIndexManager().getBufferStrategy()))
+ .getStoreState();
+ }
+
// @Override
// public Future<Void> dropZookeeperConnection() throws IOException {
//
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3ChangeLeader.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3ChangeLeader.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3ChangeLeader.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -1,34 +1,21 @@
-/**
-
-Copyright (C) SYSTAP, LLC 2006-2013. All rights reserved.
-
-Contact:
- SYSTAP, LLC
- 4501 Tower Road
- Greensboro, NC 27410
- lic...@bi...
-
-This program is free software; you can redistribute it and/or modify
-it under the terms of the GNU General Public License as published by
-the Free Software Foundation; version 2 of the License.
-
-This program is distributed in the hope that it will be useful,
-but WITHOUT ANY WARRANTY; without even the implied warranty of
-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-GNU General Public License for more details.
-
-You should have received a copy of the GNU General Public License
-along with this program; if not, write to the Free Software
-Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-*/
package com.bigdata.journal.jini.ha;
+import java.util.Random;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import com.bigdata.ha.HAGlue;
+import com.bigdata.journal.jini.ha.HAJournalTest.HAGlueTest;
public class TestHA3ChangeLeader extends AbstractHA3JournalServerTestCase {
+ public TestHA3ChangeLeader() {
+ }
+
+ public TestHA3ChangeLeader(String name) {
+ super(name);
+ }
+
/**
* We have seen problems with updates when the leader changes, this test reconstructs
* this simple scenario, with and update transaction, change of leader and then a
@@ -38,6 +25,32 @@
*/
public void testStartABC_ChangeLeader() throws Exception {
+ doStartABC_ChangeLeader(1);
+ }
+
+ public void testStartABC_ChangeLeader_2Trans() throws Exception {
+
+ doStartABC_ChangeLeader(2);
+ }
+
+ public void testStartABC_ChangeLeader_3Trans() throws Exception {
+
+ doStartABC_ChangeLeader(3);
+ }
+
+ public void testStartABC_ChangeLeader_RandomTrans() throws Exception {
+ final Random r = new Random();
+ final int ntrans = r.nextInt(900);
+ try {
+ doStartABC_ChangeLeader(ntrans);
+ } catch (Exception e) {
+ log.error("Problem with " + ntrans + " transactions");
+ throw e;
+ }
+ }
+
+ public void doStartABC_ChangeLeader(final int ntrans) throws Exception {
+
// Start 3 services
final HAGlue serverA = startA();
final HAGlue serverB = startB();
@@ -53,19 +66,131 @@
awaitCommitCounter(1L, new HAGlue[] { serverA, serverB, serverC });
/*
- * Now go through a commit point with a met quorum. The HALog
- * files should be retained at that commit point.
+ * Now go through sevearl commit points with a met quorum. The HALog
+ * files should be retained at the final commit point.
*/
- simpleTransaction();
+ for (int t = 0; t < ntrans; t++) {
+ simpleTransaction();
+ }
shutdownA();
final long token2 = awaitNextQuorumMeet(token1);
+
+ // let's commit several transactions with the new leader
+ for (int t = 0; t < 20; t++) {
+ simpleTransaction();
+ // Check store states
+ assertStoreStates(new HAGlue[] { serverB, serverC });
+ }
+
+ // And again verify binary equality of ALL journals.
+ assertDigestsEquals(new HAGlue[] { serverB, serverC });
+
+ assertTrue(token2 == quorum.token());
+ }
+
+ public void testStartABC_KillLeader_RandomTrans() throws Exception {
+ fail("Test disabled pending reconcilation of socket ticket");
+ final Random r = new Random();
+ final int ntrans = r.nextInt(900);
+ try {
+ doStartABC_KillLeader(ntrans);
+ } catch (Exception e) {
+ log.error("Problem with " + ntrans + " transactions");
+ throw e;
+ }
+ }
+
+ private void doStartABC_KillLeader(final int ntrans) throws Exception {
+
+ // Start 3 services
+ final HAGlue serverA = startA();
+ final HAGlue serverB = startB();
+ final HAGlue serverC = startC();
+
+ // Wait for a quorum meet.
+ final long token1 = awaitFullyMetQuorum();
+
+ // await pipeline
+ awaitPipeline(20, TimeUnit.SECONDS, new HAGlue[] { serverA, serverB,
+ serverC });
+
+ awaitCommitCounter(1L, new HAGlue[] { serverA, serverB, serverC });
+
+ /*
+ * Now go through sevearl commit points with a met quorum. The HALog
+ * files should be retained at the final commit point.
+ */
+ for (int t = 0; t < ntrans; t++) {
+ simpleTransaction();
+
+ // Check store states
+ assertStoreStates(new HAGlueTest[] { (HAGlueTest) serverA, (HAGlueTest) serverB, (HAGlueTest) serverC });
+ }
+
+ kill(serverA);
+
+ final long token2 = awaitNextQuorumMeet(token1);
+ // let's commit several transactions with the new leader
+ for (int t = 0; t < 20; t++) {
+ simpleTransaction();
+ }
+
+ // And again verify binary equality of ALL journals.
+ assertDigestsEquals(new HAGlue[] { serverB, serverC });
+
+ assertStoreStates(new HAGlueTest[] { (HAGlueTest) serverB, (HAGlueTest) serverC });
+ assertTrue(token2 == quorum.token());
+ }
+ /**
+ * Similar to ChangeLeader but with a LargeLoad
+ */
+ public void _testStartABC_StressChangeLeader() throws Exception {
+
+ // Start 3 services
+ final HAGlue serverA = startA();
+ final HAGlue serverB = startB();
+ final HAGlue serverC = startC();
+
+ // Wait for a quorum meet.
+ final long token1 = awaitFullyMetQuorum();
+
+ // await pipeline
+ awaitPipeline(20, TimeUnit.SECONDS, new HAGlue[] { serverA, serverB,
+ serverC });
+
+ awaitCommitCounter(1L, new HAGlue[] { serverA, serverB, serverC });
+
+ /*
+ * LOAD data on leader.
+ */
+ for (int i = 0; i < 100; i++) {
+ final FutureTask<Void> ft = new FutureTask<Void>(new LargeLoadTask(
+ token1, true/* reallyLargeLoad */, false/*dropAll*/));
+
+ // Start LOAD.
+ executorService.submit(ft);
+
+ // Await LOAD, but with a timeout.
+ ft.get(longLoadTimeoutMillis, TimeUnit.MILLISECONDS);
+ }
+
+ assertStoreStates(new HAGlue[] { serverA, serverB, serverC });
+
+ shutdownA();
+
+ final long token2 = awaitNextQuorumMeet(token1);
+
simpleTransaction();
// And again verify binary equality of ALL journals.
assertDigestsEquals(new HAGlue[] { serverB, serverC });
+
+ assertStoreStates(new HAGlue[] { serverB, serverC });
+
+ assertTrue(token2 == quorum.token());
}
}
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java 2013-11-27 14:24:34 UTC (rev 7600)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java 2013-12-03 17:58:35 UTC (rev 7601)
@@ -32,6 +32,12 @@
import com.bigdata.ha.HAGlue;
import com.bigdata.ha.HAStatusEnum;
+/**
+ * FIXME This test suite has known limitations and the utility class that it
+ * tests needs a code review and revision.
+ *
+ * @author <a href="mailto:tho...@us...">Bryan Thompson</a>
+ */
public class TestHA3DumpLogs extends AbstractHA3JournalServerTestCase {
@Override
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
