You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2018/03/29 23:04:49 UTC
[GitHub] merlimat closed pull request #1292: Avoid acquiring closeLock.readLock() on every add/read operation
merlimat closed pull request #1292: Avoid acquiring closeLock.readLock() on every add/read operation
URL: https://github.com/apache/bookkeeper/pull/1292
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
index 0bd9fe8e1..285a06aa7 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
@@ -21,6 +21,7 @@
package org.apache.bookkeeper.proto;
import static com.google.common.base.Charsets.UTF_8;
+import static org.apache.bookkeeper.util.SafeRunnable.safeRun;
import com.google.common.collect.Lists;
import com.google.protobuf.ExtensionRegistry;
@@ -48,6 +49,7 @@
import org.apache.bookkeeper.auth.ClientAuthProvider;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo;
+import org.apache.bookkeeper.common.util.SafeRunnable;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
@@ -62,7 +64,6 @@
import org.apache.bookkeeper.tls.SecurityHandlerFactory;
import org.apache.bookkeeper.util.ByteBufList;
import org.apache.bookkeeper.util.OrderedSafeExecutor;
-import org.apache.bookkeeper.util.SafeRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -113,12 +114,9 @@ public BookieClient(ClientConfiguration conf, EventLoopGroup eventLoopGroup,
this.scheduler = scheduler;
if (conf.getAddEntryTimeout() > 0 || conf.getReadEntryTimeout() > 0) {
- SafeRunnable monitor = new SafeRunnable() {
- @Override
- public void safeRun() {
- monitorPendingOperations();
- }
- };
+ SafeRunnable monitor = safeRun(() -> {
+ monitorPendingOperations();
+ });
this.timeoutFuture = this.scheduler.scheduleAtFixedRate(monitor,
conf.getTimeoutMonitorIntervalSec(),
conf.getTimeoutMonitorIntervalSec(),
@@ -189,40 +187,29 @@ public PerChannelBookieClientPool lookupClient(BookieSocketAddress addr) {
public void writeLac(final BookieSocketAddress addr, final long ledgerId, final byte[] masterKey,
final long lac, final ByteBufList toSend, final WriteLacCallback cb, final Object ctx) {
- closeLock.readLock().lock();
- try {
- final PerChannelBookieClientPool client = lookupClient(addr);
- if (client == null) {
- cb.writeLacComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
- ledgerId, addr, ctx);
- return;
- }
-
- toSend.retain();
- client.obtain(new GenericCallback<PerChannelBookieClient>() {
- @Override
- public void operationComplete(final int rc, PerChannelBookieClient pcbc) {
- if (rc != BKException.Code.OK) {
- try {
- executor.submitOrdered(ledgerId, new SafeRunnable() {
- @Override
- public void safeRun() {
- cb.writeLacComplete(rc, ledgerId, addr, ctx);
- }
- });
- } catch (RejectedExecutionException re) {
- cb.writeLacComplete(getRc(BKException.Code.InterruptedException), ledgerId, addr, ctx);
- }
- } else {
- pcbc.writeLac(ledgerId, masterKey, lac, toSend, cb, ctx);
- }
+ final PerChannelBookieClientPool client = lookupClient(addr);
+ if (client == null) {
+ cb.writeLacComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
+ ledgerId, addr, ctx);
+ return;
+ }
- toSend.release();
+ toSend.retain();
+ client.obtain((rc, pcbc) -> {
+ if (rc != BKException.Code.OK) {
+ try {
+ executor.submitOrdered(ledgerId, safeRun(() -> {
+ cb.writeLacComplete(rc, ledgerId, addr, ctx);
+ }));
+ } catch (RejectedExecutionException re) {
+ cb.writeLacComplete(getRc(BKException.Code.InterruptedException), ledgerId, addr, ctx);
}
- }, ledgerId);
- } finally {
- closeLock.readLock().unlock();
- }
+ } else {
+ pcbc.writeLac(ledgerId, masterKey, lac, toSend, cb, ctx);
+ }
+
+ toSend.release();
+ }, ledgerId);
}
private void completeAdd(final int rc,
@@ -255,26 +242,21 @@ public void addEntry(final BookieSocketAddress addr,
final WriteCallback cb,
final Object ctx,
final int options) {
- closeLock.readLock().lock();
- try {
- final PerChannelBookieClientPool client = lookupClient(addr);
- if (client == null) {
- completeAdd(getRc(BKException.Code.BookieHandleNotAvailableException),
- ledgerId, entryId, addr, cb, ctx);
- return;
- }
+ final PerChannelBookieClientPool client = lookupClient(addr);
+ if (client == null) {
+ completeAdd(getRc(BKException.Code.BookieHandleNotAvailableException),
+ ledgerId, entryId, addr, cb, ctx);
+ return;
+ }
- // Retain the buffer, since the connection could be obtained after
- // the PendingApp might have already failed
- toSend.retain();
+ // Retain the buffer, since the connection could be obtained after
+ // the PendingApp might have already failed
+ toSend.retain();
- client.obtain(ChannelReadyForAddEntryCallback.create(
- this, toSend, ledgerId, entryId, addr,
- ctx, cb, options, masterKey),
- ledgerId);
- } finally {
- closeLock.readLock().unlock();
- }
+ client.obtain(ChannelReadyForAddEntryCallback.create(
+ this, toSend, ledgerId, entryId, addr,
+ ctx, cb, options, masterKey),
+ ledgerId);
}
private void completeRead(final int rc,
@@ -371,37 +353,26 @@ public void recycle() {
public void readLac(final BookieSocketAddress addr, final long ledgerId, final ReadLacCallback cb,
final Object ctx) {
- closeLock.readLock().lock();
- try {
- final PerChannelBookieClientPool client = lookupClient(addr);
- if (client == null) {
- cb.readLacComplete(getRc(BKException.Code.BookieHandleNotAvailableException), ledgerId, null, null,
- ctx);
- return;
- }
- client.obtain(new GenericCallback<PerChannelBookieClient>() {
- @Override
- public void operationComplete(final int rc, PerChannelBookieClient pcbc) {
- if (rc != BKException.Code.OK) {
- try {
- executor.submitOrdered(ledgerId, new SafeRunnable() {
- @Override
- public void safeRun() {
- cb.readLacComplete(rc, ledgerId, null, null, ctx);
- }
- });
- } catch (RejectedExecutionException re) {
- cb.readLacComplete(getRc(BKException.Code.InterruptedException),
- ledgerId, null, null, ctx);
- }
- return;
- }
- pcbc.readLac(ledgerId, cb, ctx);
- }
- }, ledgerId);
- } finally {
- closeLock.readLock().unlock();
+ final PerChannelBookieClientPool client = lookupClient(addr);
+ if (client == null) {
+ cb.readLacComplete(getRc(BKException.Code.BookieHandleNotAvailableException), ledgerId, null, null,
+ ctx);
+ return;
}
+ client.obtain((rc, pcbc) -> {
+ if (rc != BKException.Code.OK) {
+ try {
+ executor.submitOrdered(ledgerId, safeRun(() -> {
+ cb.readLacComplete(rc, ledgerId, null, null, ctx);
+ }));
+ } catch (RejectedExecutionException re) {
+ cb.readLacComplete(getRc(BKException.Code.InterruptedException),
+ ledgerId, null, null, ctx);
+ }
+ } else {
+ pcbc.readLac(ledgerId, cb, ctx);
+ }
+ }, ledgerId);
}
public void readEntry(BookieSocketAddress addr, long ledgerId, long entryId,
@@ -411,28 +382,20 @@ public void readEntry(BookieSocketAddress addr, long ledgerId, long entryId,
public void readEntry(final BookieSocketAddress addr, final long ledgerId, final long entryId,
final ReadEntryCallback cb, final Object ctx, int flags, byte[] masterKey) {
- closeLock.readLock().lock();
- try {
- final PerChannelBookieClientPool client = lookupClient(addr);
- if (client == null) {
- cb.readEntryComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
- ledgerId, entryId, null, ctx);
- return;
- }
-
- client.obtain(new GenericCallback<PerChannelBookieClient>() {
- @Override
- public void operationComplete(final int rc, PerChannelBookieClient pcbc) {
- if (rc != BKException.Code.OK) {
- completeRead(rc, ledgerId, entryId, null, cb, ctx);
- return;
- }
- pcbc.readEntry(ledgerId, entryId, cb, ctx, flags, masterKey);
- }
- }, ledgerId);
- } finally {
- closeLock.readLock().unlock();
+ final PerChannelBookieClientPool client = lookupClient(addr);
+ if (client == null) {
+ cb.readEntryComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
+ ledgerId, entryId, null, ctx);
+ return;
}
+
+ client.obtain((rc, pcbc) -> {
+ if (rc != BKException.Code.OK) {
+ completeRead(rc, ledgerId, entryId, null, cb, ctx);
+ } else {
+ pcbc.readEntry(ledgerId, entryId, cb, ctx, flags, masterKey);
+ }
+ }, ledgerId);
}
@@ -444,65 +407,45 @@ public void readEntryWaitForLACUpdate(final BookieSocketAddress addr,
final boolean piggyBackEntry,
final ReadEntryCallback cb,
final Object ctx) {
- closeLock.readLock().lock();
- try {
- final PerChannelBookieClientPool client = lookupClient(addr);
- if (client == null) {
- completeRead(BKException.Code.BookieHandleNotAvailableException,
- ledgerId, entryId, null, cb, ctx);
- return;
- }
-
- client.obtain(new GenericCallback<PerChannelBookieClient>() {
- @Override
- public void operationComplete(final int rc, PerChannelBookieClient pcbc) {
-
- if (rc != BKException.Code.OK) {
- completeRead(rc, ledgerId, entryId, null, cb, ctx);
- return;
- }
- pcbc.readEntryWaitForLACUpdate(ledgerId, entryId, previousLAC, timeOutInMillis, piggyBackEntry, cb,
- ctx);
- }
- }, ledgerId);
- } finally {
- closeLock.readLock().unlock();
+ final PerChannelBookieClientPool client = lookupClient(addr);
+ if (client == null) {
+ completeRead(BKException.Code.BookieHandleNotAvailableException,
+ ledgerId, entryId, null, cb, ctx);
+ return;
}
+
+ client.obtain((rc, pcbc) -> {
+ if (rc != BKException.Code.OK) {
+ completeRead(rc, ledgerId, entryId, null, cb, ctx);
+ } else {
+ pcbc.readEntryWaitForLACUpdate(ledgerId, entryId, previousLAC, timeOutInMillis, piggyBackEntry, cb,
+ ctx);
+ }
+ }, ledgerId);
}
public void getBookieInfo(final BookieSocketAddress addr, final long requested, final GetBookieInfoCallback cb,
final Object ctx) {
- closeLock.readLock().lock();
- try {
- final PerChannelBookieClientPool client = lookupClient(addr);
- if (client == null) {
- cb.getBookieInfoComplete(getRc(BKException.Code.BookieHandleNotAvailableException), new BookieInfo(),
- ctx);
- return;
- }
- client.obtain(new GenericCallback<PerChannelBookieClient>() {
- @Override
- public void operationComplete(final int rc, PerChannelBookieClient pcbc) {
- if (rc != BKException.Code.OK) {
- try {
- executor.submit(new SafeRunnable() {
- @Override
- public void safeRun() {
- cb.getBookieInfoComplete(rc, new BookieInfo(), ctx);
- }
- });
- } catch (RejectedExecutionException re) {
- cb.getBookieInfoComplete(getRc(BKException.Code.InterruptedException),
- new BookieInfo(), ctx);
- }
- return;
- }
- pcbc.getBookieInfo(requested, cb, ctx);
- }
- }, requested);
- } finally {
- closeLock.readLock().unlock();
+ final PerChannelBookieClientPool client = lookupClient(addr);
+ if (client == null) {
+ cb.getBookieInfoComplete(getRc(BKException.Code.BookieHandleNotAvailableException), new BookieInfo(),
+ ctx);
+ return;
}
+ client.obtain((rc, pcbc) -> {
+ if (rc != BKException.Code.OK) {
+ try {
+ executor.submit(safeRun(() -> {
+ cb.getBookieInfoComplete(rc, new BookieInfo(), ctx);
+ }));
+ } catch (RejectedExecutionException re) {
+ cb.getBookieInfoComplete(getRc(BKException.Code.InterruptedException),
+ new BookieInfo(), ctx);
+ }
+ } else {
+ pcbc.getBookieInfo(requested, cb, ctx);
+ }
+ }, requested);
}
private void monitorPendingOperations() {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services