You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by fp...@apache.org on 2012/12/28 12:32:16 UTC
svn commit: r1426461 - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-benchmark/
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/
bookkeeper-server/src/test/java/org/apache/bookke...
Author: fpj
Date: Fri Dec 28 11:32:15 2012
New Revision: 1426461
URL: http://svn.apache.org/viewvc?rev=1426461&view=rev
Log:
BOOKKEEPER-514: TestDeadLock hanging sometimes (ivank, sijie via fpj)
Removed:
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieThrottleTest.java
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSpeculativeRead.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieReadWriteTest.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/TestDeadlock.java
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Fri Dec 28 11:32:15 2012
@@ -155,6 +155,8 @@ Trunk (unreleased changes)
BOOKKEEPER-447: Bookie can fail to recover if index pages flushed before ledger flush acknowledged (ivank via sijie)
BOOKKEEPER-520: BookieFailureTest hangs on precommit build (sijie via fpj, jira reopened)
+ BOOKKEEPER-514: TestDeadLock hanging sometimes (ivank, sijie via fpj)
+
hedwig-protocol:
BOOKKEEPER-394: CompositeException message is not useful (Stu Hood via sijie)
Modified: zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml Fri Dec 28 11:32:15 2012
@@ -67,27 +67,6 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.6.4</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- <version>0.23.1</version>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-hdfs</artifactId>
- <version>0.23.1</version>
- <scope>compile</scope>
- <!-- commons-daemon has a malformed pom, which can cause the build to fail in an
- environment which uses an artifactory cache. -->
- <exclusions>
- <exclusion>
- <groupId>commons-daemon</groupId>
- <artifactId>commons-daemon</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
@@ -113,8 +92,8 @@
<groupId>org.apache.bookkeeper</groupId>
<artifactId>bookkeeper-server</artifactId>
<version>${project.parent.version}</version>
- <type>jar</type>
<scope>compile</scope>
+ <type>jar</type>
</dependency>
<dependency>
<groupId>org.apache.bookkeeper</groupId>
@@ -133,7 +112,6 @@
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.15</version>
- <scope>provided</scope>
<exclusions>
<exclusion>
<groupId>javax.mail</groupId>
@@ -158,5 +136,25 @@
<artifactId>commons-cli</artifactId>
<version>1.2</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>0.23.1</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <version>0.23.1</version>
+ <scope>compile</scope>
+ <!-- commons-daemon has a malformed pom, which can cause the build to fail in an
+ environment which uses an artifactory cache. -->
+ <exclusions>
+ <exclusion>
+ <groupId>commons-daemon</groupId>
+ <artifactId>commons-daemon</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
</project>
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java Fri Dec 28 11:32:15 2012
@@ -28,7 +28,6 @@ import java.util.Arrays;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.Queue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.client.AsyncCallback.ReadLastConfirmedCallback;
@@ -46,6 +45,8 @@ import org.apache.bookkeeper.proto.Booki
import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.State;
import org.apache.bookkeeper.util.SafeRunnable;
+import com.google.common.util.concurrent.RateLimiter;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,8 +69,7 @@ public class LedgerHandle {
final DigestManager macManager;
final DistributionSchedule distributionSchedule;
- final Semaphore opCounterSem;
- private final Integer throttling;
+ final RateLimiter throttler;
/**
* Invalid entry id. This value is returned from methods which
@@ -96,8 +96,7 @@ public class LedgerHandle {
this.ledgerId = ledgerId;
- this.throttling = bk.getConf().getThrottleValue();
- this.opCounterSem = new Semaphore(throttling);
+ this.throttler = RateLimiter.create(bk.getConf().getThrottleValue());
macManager = DigestManager.instantiate(ledgerId, password, digestType);
this.ledgerKey = MacDigestManager.genDigest("ledger", password);
@@ -168,15 +167,6 @@ public class LedgerHandle {
}
/**
- * Return total number of available slots.
- *
- * @return int available slots
- */
- Semaphore getAvailablePermits() {
- return this.opCounterSem;
- }
-
- /**
* Add to the length of the ledger in bytes.
*
* @param delta
@@ -489,12 +479,7 @@ public class LedgerHandle {
"Invalid values for offset("+offset
+") or length("+length+")");
}
- try {
- opCounterSem.acquire();
- } catch (InterruptedException e) {
- cb.addComplete(BKException.Code.InterruptedException,
- LedgerHandle.this, INVALID_ENTRY_ID, ctx);
- }
+ throttler.acquire();
final long entryId;
final long currentLength;
@@ -504,7 +489,6 @@ public class LedgerHandle {
// updating lastAddPushed
if (metadata.isClosed()) {
LOG.warn("Attempt to add to closed ledger: " + ledgerId);
- LedgerHandle.this.opCounterSem.release();
cb.addComplete(BKException.Code.LedgerClosedException,
LedgerHandle.this, INVALID_ENTRY_ID, ctx);
return;
@@ -526,7 +510,6 @@ public class LedgerHandle {
}
});
} catch (RuntimeException e) {
- opCounterSem.release();
cb.addComplete(BKException.Code.InterruptedException,
LedgerHandle.this, INVALID_ENTRY_ID, ctx);
}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java Fri Dec 28 11:32:15 2012
@@ -168,7 +168,6 @@ class PendingAddOp implements WriteCallb
void submitCallback(final int rc) {
cb.addComplete(rc, lh, entryId, ctx);
- lh.opCounterSem.release();
}
@Override
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java Fri Dec 28 11:32:15 2012
@@ -308,7 +308,7 @@ class PendingReadOp implements Enumerati
}
void sendReadTo(InetSocketAddress to, LedgerEntryRequest entry) throws InterruptedException {
- lh.opCounterSem.acquire();
+ lh.throttler.acquire();
lh.bk.bookieClient.readEntry(to, lh.ledgerId, entry.entryId,
this, new ReadContext(to, entry));
@@ -319,8 +319,6 @@ class PendingReadOp implements Enumerati
final ReadContext rctx = (ReadContext)ctx;
final LedgerEntryRequest entry = rctx.entry;
- lh.opCounterSem.release();
-
if (rc != BKException.Code.OK) {
entry.logErrorAndReattemptRead(rctx.to, "Error: " + BKException.getMessage(rc), rc);
return;
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java Fri Dec 28 11:32:15 2012
@@ -75,7 +75,6 @@ public class PerChannelBookieClient exte
public static final int MAX_FRAME_LENGTH = 2 * 1024 * 1024; // 2M
InetSocketAddress addr;
- Semaphore opCounterSem = new Semaphore(2000);
AtomicLong totalBytesOutstanding;
ClientSocketChannelFactory channelFactory;
OrderedSafeExecutor executor;
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSpeculativeRead.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSpeculativeRead.java?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSpeculativeRead.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSpeculativeRead.java Fri Dec 28 11:32:15 2012
@@ -291,6 +291,7 @@ public class TestSpeculativeRead extends
Set<InetSocketAddress> noHost = new HashSet();
Set<InetSocketAddress> secondHostOnly = new HashSet();
secondHostOnly.add(ensemble.get(1));
+ PendingReadOp.LedgerEntryRequest req0 = null, req2 = null, req4 = null;
try {
LatchCallback latch0 = new LatchCallback();
PendingReadOp op = new PendingReadOp(l, bkspec.scheduler,
@@ -298,8 +299,7 @@ public class TestSpeculativeRead extends
// if we've already heard from all hosts,
// we only send the initial read
- PendingReadOp.LedgerEntryRequest req0
- = op.new LedgerEntryRequest(ensemble, l.getId(), 0);
+ req0 = op.new LedgerEntryRequest(ensemble, l.getId(), 0);
assertTrue("Should have sent to first",
req0.maybeSendSpeculativeRead(allHosts).equals(ensemble.get(0)));
assertNull("Should not have sent another",
@@ -307,8 +307,7 @@ public class TestSpeculativeRead extends
// if we have heard from some hosts, but not one we have sent to
// send again
- PendingReadOp.LedgerEntryRequest req2
- = op.new LedgerEntryRequest(ensemble, l.getId(), 2);
+ req2 = op.new LedgerEntryRequest(ensemble, l.getId(), 2);
assertTrue("Should have sent to third",
req2.maybeSendSpeculativeRead(noHost).equals(ensemble.get(2)));
assertTrue("Should have sent to first",
@@ -316,15 +315,25 @@ public class TestSpeculativeRead extends
// if we have heard from some hosts, which includes one we sent to
// do not read again
- PendingReadOp.LedgerEntryRequest req4
- = op.new LedgerEntryRequest(ensemble, l.getId(), 4);
+ req4 = op.new LedgerEntryRequest(ensemble, l.getId(), 4);
assertTrue("Should have sent to second",
req4.maybeSendSpeculativeRead(noHost).equals(ensemble.get(1)));
assertNull("Should not have sent another",
req4.maybeSendSpeculativeRead(secondHostOnly));
} finally {
- // wait for all ops to complete
- l.opCounterSem.acquire(bkspec.getConf().getThrottleValue());
+ for (PendingReadOp.LedgerEntryRequest req
+ : new PendingReadOp.LedgerEntryRequest[] { req0, req2, req4 }) {
+ if (req != null) {
+ int i = 0;
+ while (!req.isComplete()) {
+ if (i++ > 10) {
+ break; // wait for up to 10 seconds
+ }
+ Thread.sleep(1000);
+ }
+ assertTrue("Request should be done", req0.isComplete());
+ }
+ }
l.close();
bkspec.close();
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieReadWriteTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieReadWriteTest.java?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieReadWriteTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieReadWriteTest.java Fri Dec 28 11:32:15 2012
@@ -415,119 +415,6 @@ public class BookieReadWriteTest extends
}
}
- /**
- * Method for obtaining the available permits of a ledger handle
- * using reflection to avoid adding a new public method to the
- * class.
- *
- * @param lh
- * @return
- */
- @SuppressWarnings("unchecked")
- int getAvailablePermits(LedgerHandle lh) throws
- NoSuchFieldException, IllegalAccessException
- {
- Field field = LedgerHandle.class.getDeclaredField("opCounterSem");
- field.setAccessible(true);
- return ((Semaphore)field.get(lh)).availablePermits();
- }
-
- @Test
- public void testReadWriteAsyncSingleClientThrottle() throws
- IOException, NoSuchFieldException, IllegalAccessException {
- SyncObj sync = new SyncObj();
- try {
-
- Integer throttle = 100;
- ThrottleTestCallback tcb = new ThrottleTestCallback(throttle);
- // Create a ledger
- bkc.getConf().setThrottleValue(throttle);
- lh = bkc.createLedger(digestType, ledgerPassword);
- // bkc.initMessageDigest("SHA1");
- ledgerId = lh.getId();
- LOG.info("Ledger ID: " + lh.getId());
-
- numEntriesToWrite = 8000;
- for (int i = 0; i < (numEntriesToWrite - 2000); i++) {
- ByteBuffer entry = ByteBuffer.allocate(4);
- entry.putInt(rng.nextInt(maxInt));
- entry.position(0);
-
- entries.add(entry.array());
- entriesSize.add(entry.array().length);
- lh.asyncAddEntry(entry.array(), this, sync);
- /*
- * Check that the difference is no larger than the throttling threshold
- */
- int testValue = getAvailablePermits(lh);
- assertTrue("Difference is incorrect : " + i + ", " + sync.counter + ", " + testValue, testValue <= throttle);
- }
-
-
- for (int i = 0; i < 2000; i++) {
- ByteBuffer entry = ByteBuffer.allocate(4);
- entry.putInt(rng.nextInt(maxInt));
- entry.position(0);
-
- entries.add(entry.array());
- entriesSize.add(entry.array().length);
- lh.asyncAddEntry(entry.array(), this, sync);
-
- /*
- * Check that the difference is no larger than the throttling threshold
- */
- int testValue = getAvailablePermits(lh);
- assertTrue("Difference is incorrect : " + i + ", " + sync.counter + ", " + testValue, testValue <= throttle);
- }
-
- // wait for all entries to be acknowledged
- synchronized (sync) {
- while (sync.counter < numEntriesToWrite) {
- LOG.debug("Entries counter = " + sync.counter);
- sync.wait();
- }
- assertEquals("Error adding", BKException.Code.OK, sync.getReturnCode());
- }
-
- LOG.debug("*** WRITE COMPLETE ***");
- // close ledger
- lh.close();
-
- // *** WRITING PART COMPLETE // READ PART BEGINS ***
-
- // open ledger
- lh = bkc.openLedger(ledgerId, digestType, ledgerPassword);
- LOG.debug("Number of entries written: " + (lh.getLastAddConfirmed() + 1));
- assertTrue("Verifying number of entries written", lh.getLastAddConfirmed() == (numEntriesToWrite - 1));
-
- // read entries
- sync.counter = 0;
- for (int i = 0; i < numEntriesToWrite; i+=throttle) {
- lh.asyncReadEntries(i, i + throttle - 1, tcb, sync);
- int testValue = getAvailablePermits(lh);
- assertTrue("Difference is incorrect : " + i + ", " + sync.counter + ", " + testValue, testValue <= throttle);
- }
-
- synchronized (sync) {
- while (sync.counter < numEntriesToWrite) {
- LOG.info("Entries counter = " + sync.counter);
- sync.wait();
- }
- assertEquals("Error reading", BKException.Code.OK, sync.getReturnCode());
- }
-
- LOG.debug("*** READ COMPLETE ***");
-
- lh.close();
- } catch (BKException e) {
- LOG.error("Test failed", e);
- fail("Test failed due to BookKeeper exception");
- } catch (InterruptedException e) {
- LOG.error("Test failed", e);
- fail("Test failed due to interruption");
- }
- }
-
@Test
public void testSyncReadAsyncWriteStringsSingleClient() throws IOException {
SyncObj sync = new SyncObj();
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/TestDeadlock.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/TestDeadlock.java?rev=1426461&r1=1426460&r2=1426461&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/TestDeadlock.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/TestDeadlock.java Fri Dec 28 11:32:15 2012
@@ -141,7 +141,10 @@ public class TestDeadlock extends Hedwig
publisher.asyncPublish(topic, getMsg(9999), new TestCallback(queue), null);
}
for (int i=0; i<3; i++) {
- assertTrue(queue.take());
+ if (!queue.take()) {
+ logger.error("Error publishing to topic {}", topic);
+ ConcurrencyUtils.put(consumeQueue, false);
+ }
}
} catch (Exception e) {
logger.error("Failed to publish message to obtain permit.");