You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by "merlimat (via GitHub)" <gi...@apache.org> on 2023/03/09 07:15:08 UTC
[GitHub] [bookkeeper] merlimat commented on a diff in pull request #3848: Improve group and flush add-responses after journal sync
merlimat commented on code in PR #3848:
URL: https://github.com/apache/bookkeeper/pull/3848#discussion_r1130564777
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java:
##########
@@ -503,17 +513,26 @@ public void run() {
// responses
for (int i = 0; i < requestsCount; i++) {
ForceWriteRequest req = localRequests.get(i);
+ req.getForceWriteWaiters().forEach(ele -> {
Review Comment:
`req.process()` is already iterating over all the entries for each `ForceWriteRequest`. It would be better to reuse the same loop.
Additionally using `forEach` has 2 problems:
1. allocates an iterator, unlike doing `for (int i =0; i... .)`
2. the lambda is also capturing variables and thus it will allocate a temporary object
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java:
##########
@@ -495,6 +504,7 @@ public void run() {
journalStats.getForceWriteQueueSize().addCount(-requestsCount);
+ Set<BookieRequestHandler> writeHandlers = new HashSet<>();
Review Comment:
Using the set has its own overhead too. In this case it will allocate a `Node` object for the linked list for each entry added here.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java:
##########
@@ -495,6 +504,7 @@ public void run() {
journalStats.getForceWriteQueueSize().addCount(-requestsCount);
+ Set<BookieRequestHandler> writeHandlers = new HashSet<>();
Review Comment:
Perhaps we could use `ObjectHashSet` from carrotsearch https://github.com/carrotsearch/hppc
It's already used in Pulsar too.
We should also reuse the same set each time.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieRequestHandler.java:
##########
@@ -92,31 +92,22 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
public synchronized void prepareSendResponseV2(int rc, BookieProtocol.ParsedAddRequest req) {
if (pendingSendResponses == null) {
- pendingSendResponses = ctx.alloc().directBuffer(maxPendingResponsesSize != 0
- ? maxPendingResponsesSize : 256);
+ pendingSendResponses = ctx().alloc().directBuffer(maxPendingResponsesSize);
}
-
BookieProtoEncoding.ResponseEnDeCoderPreV3.serializeAddResponseInto(rc, req, pendingSendResponses);
}
- @Override
- public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
- if (evt == EVENT_FLUSH_ALL_PENDING_RESPONSES) {
- synchronized (this) {
- if (pendingSendResponses != null) {
- maxPendingResponsesSize = Math.max(maxPendingResponsesSize,
- pendingSendResponses.readableBytes());
- if (ctx.channel().isActive()) {
- ctx.writeAndFlush(pendingSendResponses, ctx.voidPromise());
- } else {
- pendingSendResponses.release();
- }
-
- pendingSendResponses = null;
- }
+ public synchronized void flushPendingResponse() {
+ if (pendingSendResponses != null) {
+ maxPendingResponsesSize = (int) Math.max(
+ maxPendingResponsesSize * 0.9 + 0.1 * pendingSendResponses.readableBytes(),
Review Comment:
This looks like it's going to take quite a bit of iterations to expand the cluster size. While in the meantime we have to keep expanding the `ByteBuf`, which will incur in multiple allocations and memory copies
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org