You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "void-ptr974 (via GitHub)" <gi...@apache.org> on 2026/06/14 09:29:40 UTC
[PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
void-ptr974 opened a new pull request, #26030:
URL: https://github.com/apache/pulsar/pull/26030
### Motivation
Part of #26027.
`PendingAcksMap` stores per-entry pending ack metadata as an `IntIntPair` containing the remaining unacked count and sticky key hash. That keeps one value object per pending ack entry. The main consumer ack paths only need the remaining unacked count after lookup or removal, so returning an `IntIntPair` there also creates short-lived objects that are not needed by the caller.
### Modifications
- Store the pending ack value as a packed `long` inside the existing `TreeMap<Long, TreeMap<Long, ...>>` structure.
- high 32 bits: remaining unacked count
- low 32 bits: sticky key hash
- Keep the existing ledger and entry ordering unchanged. This patch does not replace the inner map implementation.
- Add primitive accessors for the consumer hot paths:
- `getRemainingUnacked`
- `removeAndGetRemainingUnacked`
- Update `Consumer` to use the primitive accessors for ack owner lookup, full ack completion, and redelivery removal.
- Keep the existing `get` and `removeAndGet` methods that return `IntIntPair` for compatibility with existing callers.
- Add coverage for zero remaining count, signed sticky key hashes, integer boundaries, value-based removal, `updateRemainingUnacked`, and `removeAllUpTo` callbacks.
### Verifying this change
- `./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.service.PendingAcksMapTest --max-workers=1 -PtestRetryCount=0`
- `./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest --max-workers=1`
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#issuecomment-4846134577
@void-ptr974 for the `forEachScan`, do we know why we still have `277.6 us/op, 815.4 B/op`? Could the allocations be removed / reduced further?
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3507624844
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAckValues.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+
+/**
+ * Utility methods for storing pending ack values in primitive maps.
+ *
+ * <p>The high 32 bits contain the remaining unacked count and the low 32 bits contain the sticky key hash.
+ * A value with {@link #NOT_FOUND} as the remaining unacked count and 0 as the sticky key hash is reserved
+ * as the missing-entry sentinel.
+ */
+final class PendingAckValues {
+ /**
+ * Remaining unacked count reserved for missing entries.
+ */
+ static final int NOT_FOUND = -1;
+
+ /**
+ * Packed value reserved for missing entries in primitive maps.
+ */
+ static final long PACKED_NOT_FOUND = packUnchecked(NOT_FOUND, 0);
+
+ private static final long STICKY_KEY_HASH_MASK = 0xFFFF_FFFFL;
+
+ private PendingAckValues() {
+ }
+
+ /**
+ * Packs a pending ack's remaining unacked count and sticky key hash into one long value.
+ *
+ * @throws IllegalArgumentException if the remaining unacked count is negative
+ */
+ static long pack(int remainingUnacked, int stickyKeyHash) {
+ if (remainingUnacked < 0) {
+ throw new IllegalArgumentException("remainingUnacked must be non-negative");
+ }
+ return packUnchecked(remainingUnacked, stickyKeyHash);
+ }
+
+ /**
+ * Returns true if the packed value is the missing-entry sentinel.
+ */
+ static boolean isNotFound(long packedValue) {
+ return packedValue == PACKED_NOT_FOUND;
+ }
+
+ /**
+ * Returns the unpacked remaining unacked count and sticky key hash as an {@link IntIntPair}.
+ */
+ static IntIntPair toPair(long packedValue) {
+ return IntIntPair.of(remainingUnacked(packedValue), stickyKeyHash(packedValue));
+ }
+
+ /**
+ * Returns the remaining unacked count stored in the high 32 bits.
+ */
+ static int remainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ /**
+ * Returns the sticky key hash stored in the low 32 bits.
+ */
+ static int stickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
+
+ private static long packUnchecked(int remainingUnacked, int stickyKeyHash) {
Review Comment:
Fixed. I inlined it since there is no other usage.
##########
microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksMapPackingBenchmark.java:
##########
@@ -0,0 +1,566 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+import it.unimi.dsi.fastutil.longs.Long2LongMap;
+import it.unimi.dsi.fastutil.longs.Long2LongRBTreeMap;
+import it.unimi.dsi.fastutil.longs.Long2LongSortedMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectRBTreeMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectSortedMap;
+import it.unimi.dsi.fastutil.objects.ObjectBidirectionalIterator;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@BenchmarkMode(Mode.AverageTime)
+@Fork(1)
+@Warmup(iterations = 2, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS)
+public class PendingAcksMapPackingBenchmark {
+ private static final int PENDING_ACK_NOT_FOUND = -1;
+ private static final long STICKY_KEY_HASH_MASK = 0xFFFF_FFFFL;
+ private static final long PACKED_PENDING_ACK_NOT_FOUND = packPendingAckValueUnchecked(PENDING_ACK_NOT_FOUND, 0);
+
+ @Benchmark
+ public boolean addOrReplace(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.addOrReplace(state.ledgerIds[index], state.entryIds[index],
+ remainingUnacked(index), stickyKeyHash(index));
+ }
+
+ @Benchmark
+ public boolean containsHit(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.contains(state.ledgerIds[index], state.entryIds[index]);
+ }
+
+ @Benchmark
+ public long forEachScan(MapState state) {
+ return state.store.forEachScan();
+ }
+
+ @Benchmark
+ public IntIntPair getPairHit(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.get(state.ledgerIds[index], state.entryIds[index]);
+ }
+
+ @Benchmark
+ public int getRemainingUnackedHit(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.getRemainingUnacked(state.ledgerIds[index], state.entryIds[index]);
+ }
+
+ @Benchmark
+ public long removeAllUpToBeforeFirstEntry(MapState state) {
+ return state.store.removeAllUpTo(0, -1);
+ }
+
+ @Benchmark
+ public long removeAllUpToSmallPrefixAndRefill(PrefixRemoveState state) {
+ long removed = state.store.removeAllUpTo(0, state.prefixEntries - 1L);
+ for (int i = 0; i < state.prefixEntries; i++) {
+ state.store.addOrReplace(0, i, remainingUnacked(i), stickyKeyHash(i));
+ }
+ return removed;
+ }
+
+ @Benchmark
+ public int removeAndGetRemainingAndAdd(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ long ledgerId = state.ledgerIds[index];
+ long entryId = state.entryIds[index];
+ int remainingUnacked = state.store.removeAndGetRemainingUnacked(ledgerId, entryId);
+ state.store.addOrReplace(ledgerId, entryId,
+ remainingUnacked == PENDING_ACK_NOT_FOUND ? remainingUnacked(index) : remainingUnacked,
+ stickyKeyHash(index));
+ return remainingUnacked;
+ }
+
+ @Benchmark
+ public boolean removeWithValueAndAdd(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ long ledgerId = state.ledgerIds[index];
+ long entryId = state.entryIds[index];
+ int remainingUnacked = remainingUnacked(index);
+ int stickyKeyHash = stickyKeyHash(index);
+ boolean removed = state.store.remove(ledgerId, entryId, remainingUnacked, stickyKeyHash);
+ state.store.addOrReplace(ledgerId, entryId, remainingUnacked, stickyKeyHash);
+ return removed;
+ }
+
+ @Benchmark
+ public boolean updateRemainingUnacked(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.updateRemainingUnacked(state.ledgerIds[index], state.entryIds[index], 1);
+ }
+
+ @State(Scope.Benchmark)
+ public static class MapState {
+ @Param({"FASTUTIL_OBJECT", "FASTUTIL_PACKED", "FASTUTIL_PACKED_SENTINEL"})
+ private Implementation implementation;
+
+ @Param({"50000"})
+ private int entries;
+
+ @Param({"1"})
+ private int ledgers;
+
+ private PendingAcksStore store;
+ private long[] ledgerIds;
+ private long[] entryIds;
+
+ @Setup(Level.Trial)
+ public void setup() {
+ store = implementation.createStore();
+ ledgerIds = new long[entries];
+ entryIds = new long[entries];
+ populate(store, entries, ledgers, ledgerIds, entryIds);
+ }
+ }
+
+ @State(Scope.Thread)
+ public static class PrefixRemoveState {
+ @Param({"FASTUTIL_OBJECT", "FASTUTIL_PACKED", "FASTUTIL_PACKED_SENTINEL"})
+ private Implementation implementation;
+
+ @Param({"50000"})
+ private int entries;
+
+ private PendingAcksStore store;
+ private int prefixEntries;
+
+ @Setup(Level.Trial)
+ public void setup() {
+ store = implementation.createStore();
+ prefixEntries = Math.max(1, entries / 50);
+ populate(store, entries, 1, null, null);
+ }
+ }
+
+ @State(Scope.Thread)
+ public static class CursorState {
+ private int cursor;
+
+ int next(int bound) {
+ int next = cursor++;
+ if (cursor == bound) {
+ cursor = 0;
+ }
+ return next;
+ }
+ }
+
+ public enum Implementation {
+ FASTUTIL_OBJECT {
+ @Override
+ PendingAcksStore createStore() {
+ return new FastutilObjectStore();
+ }
+ },
+ FASTUTIL_PACKED {
+ @Override
+ PendingAcksStore createStore() {
+ return new FastutilPackedStore(false);
+ }
+ },
+ FASTUTIL_PACKED_SENTINEL {
+ @Override
+ PendingAcksStore createStore() {
+ return new FastutilPackedStore(true);
+ }
+ };
+
+ abstract PendingAcksStore createStore();
+ }
+
+ private interface PendingAcksStore {
+ boolean addOrReplace(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash);
+
+ boolean contains(long ledgerId, long entryId);
+
+ IntIntPair get(long ledgerId, long entryId);
+
+ int getRemainingUnacked(long ledgerId, long entryId);
+
+ boolean remove(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash);
+
+ int removeAndGetRemainingUnacked(long ledgerId, long entryId);
+
+ boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta);
+
+ long forEachScan();
+
+ long removeAllUpTo(long markDeleteLedgerId, long markDeleteEntryId);
+ }
+
+ private static final class FastutilObjectStore implements PendingAcksStore {
+ private final Long2ObjectSortedMap<Long2ObjectSortedMap<IntIntPair>> pendingAcks =
+ new Long2ObjectRBTreeMap<>();
+
+ @Override
+ public boolean addOrReplace(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ Long2ObjectSortedMap<IntIntPair> ledgerPendingAcks =
+ pendingAcks.computeIfAbsent(ledgerId, k -> new Long2ObjectRBTreeMap<>());
+ ledgerPendingAcks.put(entryId, IntIntPair.of(remainingUnacked, stickyKeyHash));
+ return true;
+ }
+
+ @Override
+ public boolean contains(long ledgerId, long entryId) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ return ledgerMap != null && ledgerMap.containsKey(entryId);
+ }
+
+ @Override
+ public IntIntPair get(long ledgerId, long entryId) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ return ledgerMap == null ? null : ledgerMap.get(entryId);
+ }
+
+ @Override
+ public int getRemainingUnacked(long ledgerId, long entryId) {
+ IntIntPair value = get(ledgerId, entryId);
+ return value == null ? PENDING_ACK_NOT_FOUND : value.leftInt();
+ }
+
+ @Override
+ public boolean remove(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ IntIntPair value = ledgerMap == null ? null : ledgerMap.get(entryId);
+ if (value == null || value.leftInt() != remainingUnacked || value.rightInt() != stickyKeyHash) {
+ return false;
+ }
+ ledgerMap.remove(entryId);
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return true;
+ }
+
+ @Override
+ public int removeAndGetRemainingUnacked(long ledgerId, long entryId) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ IntIntPair removed = ledgerMap.remove(entryId);
+ if (removed == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return removed.leftInt();
+ }
+
+ @Override
+ public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ IntIntPair value = ledgerMap == null ? null : ledgerMap.get(entryId);
+ if (value == null) {
+ return false;
+ }
+ int newRemaining = value.leftInt() - ackedDelta;
+ if (newRemaining < 0) {
+ return false;
+ }
+ ledgerMap.put(entryId, IntIntPair.of(newRemaining, value.rightInt()));
+ return true;
+ }
+
+ @Override
+ public long forEachScan() {
+ long sum = 0;
+ for (Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>> entry : pendingAcks.long2ObjectEntrySet()) {
+ long ledgerId = entry.getLongKey();
+ for (Long2ObjectMap.Entry<IntIntPair> pendingAckEntry : entry.getValue().long2ObjectEntrySet()) {
+ IntIntPair value = pendingAckEntry.getValue();
+ sum += ledgerId + pendingAckEntry.getLongKey() + value.leftInt() + value.rightInt();
+ }
+ }
+ return sum;
+ }
+
+ @Override
+ public long removeAllUpTo(long markDeleteLedgerId, long markDeleteEntryId) {
+ long removed = 0;
+ ObjectBidirectionalIterator<Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>>> ledgerIterator =
+ pendingAcks.headMap(markDeleteLedgerId + 1).long2ObjectEntrySet().iterator();
+ while (ledgerIterator.hasNext()) {
+ Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>> entry = ledgerIterator.next();
+ long ledgerId = entry.getLongKey();
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = entry.getValue();
+ Long2ObjectSortedMap<IntIntPair> ledgerMapHead =
+ ledgerId == markDeleteLedgerId ? ledgerMap.headMap(markDeleteEntryId + 1) : ledgerMap;
+ ObjectBidirectionalIterator<Long2ObjectMap.Entry<IntIntPair>> entryIterator =
+ ledgerMapHead.long2ObjectEntrySet().iterator();
+ while (entryIterator.hasNext()) {
+ entryIterator.next();
+ entryIterator.remove();
+ removed++;
+ }
+ if (ledgerMap.isEmpty()) {
+ ledgerIterator.remove();
+ }
+ }
+ return removed;
+ }
+
+ private void removeLedgerIfEmpty(long ledgerId, Long2ObjectSortedMap<IntIntPair> ledgerMap) {
+ if (ledgerMap.isEmpty()) {
+ pendingAcks.remove(ledgerId);
+ }
+ }
+ }
+
+ private static final class FastutilPackedStore implements PendingAcksStore {
+ private final Long2ObjectSortedMap<Long2LongSortedMap> pendingAcks = new Long2ObjectRBTreeMap<>();
+ private final boolean sentinelDefault;
+
+ private FastutilPackedStore(boolean sentinelDefault) {
+ this.sentinelDefault = sentinelDefault;
+ }
+
+ @Override
+ public boolean addOrReplace(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ Long2LongSortedMap ledgerPendingAcks = pendingAcks.computeIfAbsent(ledgerId, k -> newLedgerPendingAcks());
+ long packedValue = packPendingAckValue(remainingUnacked, stickyKeyHash);
+ ledgerPendingAcks.put(entryId, packedValue);
+ return true;
+ }
+
+ @Override
+ public boolean contains(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ return ledgerMap != null && ledgerMap.containsKey(entryId);
+ }
+
+ @Override
+ public IntIntPair get(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return null;
+ }
+ if (sentinelDefault) {
+ long packedValue = ledgerMap.get(entryId);
+ return packedValue == PACKED_PENDING_ACK_NOT_FOUND ? null : unpackPendingAckValue(packedValue);
+ }
+ if (!ledgerMap.containsKey(entryId)) {
+ return null;
+ }
+ return unpackPendingAckValue(ledgerMap.get(entryId));
+ }
+
+ @Override
+ public int getRemainingUnacked(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ if (sentinelDefault) {
+ long packedValue = ledgerMap.get(entryId);
+ return packedValue == PACKED_PENDING_ACK_NOT_FOUND
+ ? PENDING_ACK_NOT_FOUND : unpackRemainingUnacked(packedValue);
+ }
+ if (!ledgerMap.containsKey(entryId)) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ return unpackRemainingUnacked(ledgerMap.get(entryId));
+ }
+
+ @Override
+ public boolean remove(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ if (remainingUnacked < 0) {
+ return false;
+ }
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return false;
+ }
+ long expectedValue = packPendingAckValue(remainingUnacked, stickyKeyHash);
+ if (sentinelDefault) {
+ if (ledgerMap.get(entryId) != expectedValue) {
+ return false;
+ }
+ } else if (!ledgerMap.containsKey(entryId) || ledgerMap.get(entryId) != expectedValue) {
+ return false;
+ }
+ ledgerMap.remove(entryId);
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return true;
+ }
+
+ @Override
+ public int removeAndGetRemainingUnacked(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ if (sentinelDefault) {
+ long removed = ledgerMap.remove(entryId);
+ if (removed == PACKED_PENDING_ACK_NOT_FOUND) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return unpackRemainingUnacked(removed);
+ }
+ if (!ledgerMap.containsKey(entryId)) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ long removed = ledgerMap.remove(entryId);
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return unpackRemainingUnacked(removed);
+ }
+
+ @Override
+ public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return false;
+ }
+ long packedValue;
+ if (sentinelDefault) {
+ packedValue = ledgerMap.get(entryId);
+ if (packedValue == PACKED_PENDING_ACK_NOT_FOUND) {
+ return false;
+ }
+ } else {
+ if (!ledgerMap.containsKey(entryId)) {
+ return false;
+ }
+ packedValue = ledgerMap.get(entryId);
+ }
+ int newRemaining = unpackRemainingUnacked(packedValue) - ackedDelta;
+ if (newRemaining < 0) {
+ return false;
+ }
+ ledgerMap.put(entryId, packPendingAckValue(newRemaining, unpackStickyKeyHash(packedValue)));
+ return true;
+ }
+
+ @Override
+ public long forEachScan() {
+ long sum = 0;
+ for (Long2ObjectMap.Entry<Long2LongSortedMap> entry : pendingAcks.long2ObjectEntrySet()) {
+ long ledgerId = entry.getLongKey();
+ for (Long2LongMap.Entry pendingAckEntry : entry.getValue().long2LongEntrySet()) {
+ long packedValue = pendingAckEntry.getLongValue();
+ sum += ledgerId + pendingAckEntry.getLongKey()
+ + unpackRemainingUnacked(packedValue) + unpackStickyKeyHash(packedValue);
+ }
+ }
+ return sum;
+ }
Review Comment:
Fixed. The benchmark now uses an inline former implementation and delegates the current implementation directly to `PendingAcksMap`.
I also updated the PR description with the latest JMH results.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#issuecomment-4845111123
Thanks, I'll avoid force-pushing after review has started next time.
I also updated the title and PR description based on your suggestion, including the self-contained context, memory estimate, and JOL layout details.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3498817862
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -504,6 +571,28 @@ private void internalRemoveAllUpTo(long markDeleteLedgerId, long markDeleteEntry
}
}
+ private static long packPendingAckValue(int remainingUnacked, int stickyKeyHash) {
+ return ((long) remainingUnacked << Integer.SIZE) | (stickyKeyHash & STICKY_KEY_HASH_MASK);
+ }
+
+ private static Long2LongSortedMap newLedgerPendingAcks() {
+ Long2LongRBTreeMap ledgerPendingAcks = new Long2LongRBTreeMap();
+ ledgerPendingAcks.defaultReturnValue(PACKED_PENDING_ACK_NOT_FOUND);
+ return ledgerPendingAcks;
+ }
+
+ private static IntIntPair unpackPendingAckValue(long packedValue) {
+ return IntIntPair.of(unpackRemainingUnacked(packedValue), unpackStickyKeyHash(packedValue));
+ }
+
+ private static int unpackRemainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ private static int unpackStickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
Review Comment:
Good point, thanks. I extracted the packing/unpacking logic into a package-private PendingAckValue helper and added focused tests for round trips and sentinel behavior.
One detail worth discussing: PendingAckValue.pack now rejects negative remainingUnacked values because -1 is reserved as the not-found sentinel. In the current dispatch path, remainingUnacked comes from the entry batch size or ack-set cardinality, so it is not expected to be -1 in normal runtime use.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#issuecomment-4844077872
Regarding the current title "[improve][broker] Part 2: Pack PendingAcksMap values into long", please revisit it.
Something like "Optimize PendingAcksMap by using a primitive long2long map for values".
In the PR description, the calculation of memory consumption can be shown for before and after. The calculation could be based on theoretical analysis (for example when there's no IntIntPair values, there's no object reference overhead etc.). In PIP-379, I made [this type of estimation](https://github.com/apache/pulsar/blob/master/pip/pip-379.md#:~:text=The%20memory%20consumption%20estimate,hashes%3A%20about%20800%20kB). It's obviously not the exact amount that would be consumed, but a good estimate. Something similar could be explained in the justification for this PR. It could be better to make the description contain the justification since referring to "part 2" in #26027 is outdated.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3500642375
##########
microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksMapPackingBenchmark.java:
##########
@@ -0,0 +1,566 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+import it.unimi.dsi.fastutil.longs.Long2LongMap;
+import it.unimi.dsi.fastutil.longs.Long2LongRBTreeMap;
+import it.unimi.dsi.fastutil.longs.Long2LongSortedMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectRBTreeMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectSortedMap;
+import it.unimi.dsi.fastutil.objects.ObjectBidirectionalIterator;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@BenchmarkMode(Mode.AverageTime)
+@Fork(1)
+@Warmup(iterations = 2, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS)
+public class PendingAcksMapPackingBenchmark {
+ private static final int PENDING_ACK_NOT_FOUND = -1;
+ private static final long STICKY_KEY_HASH_MASK = 0xFFFF_FFFFL;
+ private static final long PACKED_PENDING_ACK_NOT_FOUND = packPendingAckValueUnchecked(PENDING_ACK_NOT_FOUND, 0);
+
+ @Benchmark
+ public boolean addOrReplace(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.addOrReplace(state.ledgerIds[index], state.entryIds[index],
+ remainingUnacked(index), stickyKeyHash(index));
+ }
+
+ @Benchmark
+ public boolean containsHit(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.contains(state.ledgerIds[index], state.entryIds[index]);
+ }
+
+ @Benchmark
+ public long forEachScan(MapState state) {
+ return state.store.forEachScan();
+ }
+
+ @Benchmark
+ public IntIntPair getPairHit(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.get(state.ledgerIds[index], state.entryIds[index]);
+ }
+
+ @Benchmark
+ public int getRemainingUnackedHit(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.getRemainingUnacked(state.ledgerIds[index], state.entryIds[index]);
+ }
+
+ @Benchmark
+ public long removeAllUpToBeforeFirstEntry(MapState state) {
+ return state.store.removeAllUpTo(0, -1);
+ }
+
+ @Benchmark
+ public long removeAllUpToSmallPrefixAndRefill(PrefixRemoveState state) {
+ long removed = state.store.removeAllUpTo(0, state.prefixEntries - 1L);
+ for (int i = 0; i < state.prefixEntries; i++) {
+ state.store.addOrReplace(0, i, remainingUnacked(i), stickyKeyHash(i));
+ }
+ return removed;
+ }
+
+ @Benchmark
+ public int removeAndGetRemainingAndAdd(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ long ledgerId = state.ledgerIds[index];
+ long entryId = state.entryIds[index];
+ int remainingUnacked = state.store.removeAndGetRemainingUnacked(ledgerId, entryId);
+ state.store.addOrReplace(ledgerId, entryId,
+ remainingUnacked == PENDING_ACK_NOT_FOUND ? remainingUnacked(index) : remainingUnacked,
+ stickyKeyHash(index));
+ return remainingUnacked;
+ }
+
+ @Benchmark
+ public boolean removeWithValueAndAdd(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ long ledgerId = state.ledgerIds[index];
+ long entryId = state.entryIds[index];
+ int remainingUnacked = remainingUnacked(index);
+ int stickyKeyHash = stickyKeyHash(index);
+ boolean removed = state.store.remove(ledgerId, entryId, remainingUnacked, stickyKeyHash);
+ state.store.addOrReplace(ledgerId, entryId, remainingUnacked, stickyKeyHash);
+ return removed;
+ }
+
+ @Benchmark
+ public boolean updateRemainingUnacked(MapState state, CursorState cursor) {
+ int index = cursor.next(state.entries);
+ return state.store.updateRemainingUnacked(state.ledgerIds[index], state.entryIds[index], 1);
+ }
+
+ @State(Scope.Benchmark)
+ public static class MapState {
+ @Param({"FASTUTIL_OBJECT", "FASTUTIL_PACKED", "FASTUTIL_PACKED_SENTINEL"})
+ private Implementation implementation;
+
+ @Param({"50000"})
+ private int entries;
+
+ @Param({"1"})
+ private int ledgers;
+
+ private PendingAcksStore store;
+ private long[] ledgerIds;
+ private long[] entryIds;
+
+ @Setup(Level.Trial)
+ public void setup() {
+ store = implementation.createStore();
+ ledgerIds = new long[entries];
+ entryIds = new long[entries];
+ populate(store, entries, ledgers, ledgerIds, entryIds);
+ }
+ }
+
+ @State(Scope.Thread)
+ public static class PrefixRemoveState {
+ @Param({"FASTUTIL_OBJECT", "FASTUTIL_PACKED", "FASTUTIL_PACKED_SENTINEL"})
+ private Implementation implementation;
+
+ @Param({"50000"})
+ private int entries;
+
+ private PendingAcksStore store;
+ private int prefixEntries;
+
+ @Setup(Level.Trial)
+ public void setup() {
+ store = implementation.createStore();
+ prefixEntries = Math.max(1, entries / 50);
+ populate(store, entries, 1, null, null);
+ }
+ }
+
+ @State(Scope.Thread)
+ public static class CursorState {
+ private int cursor;
+
+ int next(int bound) {
+ int next = cursor++;
+ if (cursor == bound) {
+ cursor = 0;
+ }
+ return next;
+ }
+ }
+
+ public enum Implementation {
+ FASTUTIL_OBJECT {
+ @Override
+ PendingAcksStore createStore() {
+ return new FastutilObjectStore();
+ }
+ },
+ FASTUTIL_PACKED {
+ @Override
+ PendingAcksStore createStore() {
+ return new FastutilPackedStore(false);
+ }
+ },
+ FASTUTIL_PACKED_SENTINEL {
+ @Override
+ PendingAcksStore createStore() {
+ return new FastutilPackedStore(true);
+ }
+ };
+
+ abstract PendingAcksStore createStore();
+ }
+
+ private interface PendingAcksStore {
+ boolean addOrReplace(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash);
+
+ boolean contains(long ledgerId, long entryId);
+
+ IntIntPair get(long ledgerId, long entryId);
+
+ int getRemainingUnacked(long ledgerId, long entryId);
+
+ boolean remove(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash);
+
+ int removeAndGetRemainingUnacked(long ledgerId, long entryId);
+
+ boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta);
+
+ long forEachScan();
+
+ long removeAllUpTo(long markDeleteLedgerId, long markDeleteEntryId);
+ }
+
+ private static final class FastutilObjectStore implements PendingAcksStore {
+ private final Long2ObjectSortedMap<Long2ObjectSortedMap<IntIntPair>> pendingAcks =
+ new Long2ObjectRBTreeMap<>();
+
+ @Override
+ public boolean addOrReplace(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ Long2ObjectSortedMap<IntIntPair> ledgerPendingAcks =
+ pendingAcks.computeIfAbsent(ledgerId, k -> new Long2ObjectRBTreeMap<>());
+ ledgerPendingAcks.put(entryId, IntIntPair.of(remainingUnacked, stickyKeyHash));
+ return true;
+ }
+
+ @Override
+ public boolean contains(long ledgerId, long entryId) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ return ledgerMap != null && ledgerMap.containsKey(entryId);
+ }
+
+ @Override
+ public IntIntPair get(long ledgerId, long entryId) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ return ledgerMap == null ? null : ledgerMap.get(entryId);
+ }
+
+ @Override
+ public int getRemainingUnacked(long ledgerId, long entryId) {
+ IntIntPair value = get(ledgerId, entryId);
+ return value == null ? PENDING_ACK_NOT_FOUND : value.leftInt();
+ }
+
+ @Override
+ public boolean remove(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ IntIntPair value = ledgerMap == null ? null : ledgerMap.get(entryId);
+ if (value == null || value.leftInt() != remainingUnacked || value.rightInt() != stickyKeyHash) {
+ return false;
+ }
+ ledgerMap.remove(entryId);
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return true;
+ }
+
+ @Override
+ public int removeAndGetRemainingUnacked(long ledgerId, long entryId) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ IntIntPair removed = ledgerMap.remove(entryId);
+ if (removed == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return removed.leftInt();
+ }
+
+ @Override
+ public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta) {
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = pendingAcks.get(ledgerId);
+ IntIntPair value = ledgerMap == null ? null : ledgerMap.get(entryId);
+ if (value == null) {
+ return false;
+ }
+ int newRemaining = value.leftInt() - ackedDelta;
+ if (newRemaining < 0) {
+ return false;
+ }
+ ledgerMap.put(entryId, IntIntPair.of(newRemaining, value.rightInt()));
+ return true;
+ }
+
+ @Override
+ public long forEachScan() {
+ long sum = 0;
+ for (Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>> entry : pendingAcks.long2ObjectEntrySet()) {
+ long ledgerId = entry.getLongKey();
+ for (Long2ObjectMap.Entry<IntIntPair> pendingAckEntry : entry.getValue().long2ObjectEntrySet()) {
+ IntIntPair value = pendingAckEntry.getValue();
+ sum += ledgerId + pendingAckEntry.getLongKey() + value.leftInt() + value.rightInt();
+ }
+ }
+ return sum;
+ }
+
+ @Override
+ public long removeAllUpTo(long markDeleteLedgerId, long markDeleteEntryId) {
+ long removed = 0;
+ ObjectBidirectionalIterator<Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>>> ledgerIterator =
+ pendingAcks.headMap(markDeleteLedgerId + 1).long2ObjectEntrySet().iterator();
+ while (ledgerIterator.hasNext()) {
+ Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>> entry = ledgerIterator.next();
+ long ledgerId = entry.getLongKey();
+ Long2ObjectSortedMap<IntIntPair> ledgerMap = entry.getValue();
+ Long2ObjectSortedMap<IntIntPair> ledgerMapHead =
+ ledgerId == markDeleteLedgerId ? ledgerMap.headMap(markDeleteEntryId + 1) : ledgerMap;
+ ObjectBidirectionalIterator<Long2ObjectMap.Entry<IntIntPair>> entryIterator =
+ ledgerMapHead.long2ObjectEntrySet().iterator();
+ while (entryIterator.hasNext()) {
+ entryIterator.next();
+ entryIterator.remove();
+ removed++;
+ }
+ if (ledgerMap.isEmpty()) {
+ ledgerIterator.remove();
+ }
+ }
+ return removed;
+ }
+
+ private void removeLedgerIfEmpty(long ledgerId, Long2ObjectSortedMap<IntIntPair> ledgerMap) {
+ if (ledgerMap.isEmpty()) {
+ pendingAcks.remove(ledgerId);
+ }
+ }
+ }
+
+ private static final class FastutilPackedStore implements PendingAcksStore {
+ private final Long2ObjectSortedMap<Long2LongSortedMap> pendingAcks = new Long2ObjectRBTreeMap<>();
+ private final boolean sentinelDefault;
+
+ private FastutilPackedStore(boolean sentinelDefault) {
+ this.sentinelDefault = sentinelDefault;
+ }
+
+ @Override
+ public boolean addOrReplace(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ Long2LongSortedMap ledgerPendingAcks = pendingAcks.computeIfAbsent(ledgerId, k -> newLedgerPendingAcks());
+ long packedValue = packPendingAckValue(remainingUnacked, stickyKeyHash);
+ ledgerPendingAcks.put(entryId, packedValue);
+ return true;
+ }
+
+ @Override
+ public boolean contains(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ return ledgerMap != null && ledgerMap.containsKey(entryId);
+ }
+
+ @Override
+ public IntIntPair get(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return null;
+ }
+ if (sentinelDefault) {
+ long packedValue = ledgerMap.get(entryId);
+ return packedValue == PACKED_PENDING_ACK_NOT_FOUND ? null : unpackPendingAckValue(packedValue);
+ }
+ if (!ledgerMap.containsKey(entryId)) {
+ return null;
+ }
+ return unpackPendingAckValue(ledgerMap.get(entryId));
+ }
+
+ @Override
+ public int getRemainingUnacked(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ if (sentinelDefault) {
+ long packedValue = ledgerMap.get(entryId);
+ return packedValue == PACKED_PENDING_ACK_NOT_FOUND
+ ? PENDING_ACK_NOT_FOUND : unpackRemainingUnacked(packedValue);
+ }
+ if (!ledgerMap.containsKey(entryId)) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ return unpackRemainingUnacked(ledgerMap.get(entryId));
+ }
+
+ @Override
+ public boolean remove(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ if (remainingUnacked < 0) {
+ return false;
+ }
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return false;
+ }
+ long expectedValue = packPendingAckValue(remainingUnacked, stickyKeyHash);
+ if (sentinelDefault) {
+ if (ledgerMap.get(entryId) != expectedValue) {
+ return false;
+ }
+ } else if (!ledgerMap.containsKey(entryId) || ledgerMap.get(entryId) != expectedValue) {
+ return false;
+ }
+ ledgerMap.remove(entryId);
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return true;
+ }
+
+ @Override
+ public int removeAndGetRemainingUnacked(long ledgerId, long entryId) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ if (sentinelDefault) {
+ long removed = ledgerMap.remove(entryId);
+ if (removed == PACKED_PENDING_ACK_NOT_FOUND) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return unpackRemainingUnacked(removed);
+ }
+ if (!ledgerMap.containsKey(entryId)) {
+ return PENDING_ACK_NOT_FOUND;
+ }
+ long removed = ledgerMap.remove(entryId);
+ removeLedgerIfEmpty(ledgerId, ledgerMap);
+ return unpackRemainingUnacked(removed);
+ }
+
+ @Override
+ public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta) {
+ Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ if (ledgerMap == null) {
+ return false;
+ }
+ long packedValue;
+ if (sentinelDefault) {
+ packedValue = ledgerMap.get(entryId);
+ if (packedValue == PACKED_PENDING_ACK_NOT_FOUND) {
+ return false;
+ }
+ } else {
+ if (!ledgerMap.containsKey(entryId)) {
+ return false;
+ }
+ packedValue = ledgerMap.get(entryId);
+ }
+ int newRemaining = unpackRemainingUnacked(packedValue) - ackedDelta;
+ if (newRemaining < 0) {
+ return false;
+ }
+ ledgerMap.put(entryId, packPendingAckValue(newRemaining, unpackStickyKeyHash(packedValue)));
+ return true;
+ }
+
+ @Override
+ public long forEachScan() {
+ long sum = 0;
+ for (Long2ObjectMap.Entry<Long2LongSortedMap> entry : pendingAcks.long2ObjectEntrySet()) {
+ long ledgerId = entry.getLongKey();
+ for (Long2LongMap.Entry pendingAckEntry : entry.getValue().long2LongEntrySet()) {
+ long packedValue = pendingAckEntry.getLongValue();
+ sum += ledgerId + pendingAckEntry.getLongKey()
+ + unpackRemainingUnacked(packedValue) + unpackStickyKeyHash(packedValue);
+ }
+ }
+ return sum;
+ }
Review Comment:
This doesn't seem to match the latest implementation. It seems that it would be better to have a benchmark where the former implementation is copied to inline classes and the current implementation is used directly.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3498817862
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -504,6 +571,28 @@ private void internalRemoveAllUpTo(long markDeleteLedgerId, long markDeleteEntry
}
}
+ private static long packPendingAckValue(int remainingUnacked, int stickyKeyHash) {
+ return ((long) remainingUnacked << Integer.SIZE) | (stickyKeyHash & STICKY_KEY_HASH_MASK);
+ }
+
+ private static Long2LongSortedMap newLedgerPendingAcks() {
+ Long2LongRBTreeMap ledgerPendingAcks = new Long2LongRBTreeMap();
+ ledgerPendingAcks.defaultReturnValue(PACKED_PENDING_ACK_NOT_FOUND);
+ return ledgerPendingAcks;
+ }
+
+ private static IntIntPair unpackPendingAckValue(long packedValue) {
+ return IntIntPair.of(unpackRemainingUnacked(packedValue), unpackStickyKeyHash(packedValue));
+ }
+
+ private static int unpackRemainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ private static int unpackStickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
Review Comment:
Good point, thanks. I extracted the packing/unpacking logic into a package-private PendingAckValue helper and added focused tests for round trips and sentinel behavior.
One detail worth discussing: PendingAckValue.pack now rejects negative remainingUnacked values because -1 is reserved as the not-found sentinel.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#issuecomment-4738355943
Thanks @lhotari. I updated this on top of #26032.
The latest version uses fastutil and keeps the scope to the value-layout change. I also updated the benchmark/JOL numbers in the PR description.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3499008802
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -131,6 +136,7 @@ public interface PendingAcksConsumer {
* @return true if the pending ack was added, and it's allowed to send a message, false otherwise
*/
public boolean addPendingAckIfAllowed(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ long packedValue = PendingAckValue.pack(remainingUnacked, stickyKeyHash);
Review Comment:
Fixed.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3500658080
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAckValues.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+
+/**
+ * Utility methods for storing pending ack values in primitive maps.
+ *
+ * <p>The high 32 bits contain the remaining unacked count and the low 32 bits contain the sticky key hash.
+ * A value with {@link #NOT_FOUND} as the remaining unacked count and 0 as the sticky key hash is reserved
+ * as the missing-entry sentinel.
+ */
+final class PendingAckValues {
+ /**
+ * Remaining unacked count reserved for missing entries.
+ */
+ static final int NOT_FOUND = -1;
+
+ /**
+ * Packed value reserved for missing entries in primitive maps.
+ */
+ static final long PACKED_NOT_FOUND = packUnchecked(NOT_FOUND, 0);
+
+ private static final long STICKY_KEY_HASH_MASK = 0xFFFF_FFFFL;
+
+ private PendingAckValues() {
+ }
+
+ /**
+ * Packs a pending ack's remaining unacked count and sticky key hash into one long value.
+ *
+ * @throws IllegalArgumentException if the remaining unacked count is negative
+ */
+ static long pack(int remainingUnacked, int stickyKeyHash) {
+ if (remainingUnacked < 0) {
+ throw new IllegalArgumentException("remainingUnacked must be non-negative");
+ }
+ return packUnchecked(remainingUnacked, stickyKeyHash);
+ }
+
+ /**
+ * Returns true if the packed value is the missing-entry sentinel.
+ */
+ static boolean isNotFound(long packedValue) {
+ return packedValue == PACKED_NOT_FOUND;
+ }
+
+ /**
+ * Returns the unpacked remaining unacked count and sticky key hash as an {@link IntIntPair}.
+ */
+ static IntIntPair toPair(long packedValue) {
+ return IntIntPair.of(remainingUnacked(packedValue), stickyKeyHash(packedValue));
+ }
+
+ /**
+ * Returns the remaining unacked count stored in the high 32 bits.
+ */
+ static int remainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ /**
+ * Returns the sticky key hash stored in the low 32 bits.
+ */
+ static int stickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
+
+ private static long packUnchecked(int remainingUnacked, int stickyKeyHash) {
Review Comment:
moving this method directly after the `pack` method would make the code easier to read.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3498817862
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -504,6 +571,28 @@ private void internalRemoveAllUpTo(long markDeleteLedgerId, long markDeleteEntry
}
}
+ private static long packPendingAckValue(int remainingUnacked, int stickyKeyHash) {
+ return ((long) remainingUnacked << Integer.SIZE) | (stickyKeyHash & STICKY_KEY_HASH_MASK);
+ }
+
+ private static Long2LongSortedMap newLedgerPendingAcks() {
+ Long2LongRBTreeMap ledgerPendingAcks = new Long2LongRBTreeMap();
+ ledgerPendingAcks.defaultReturnValue(PACKED_PENDING_ACK_NOT_FOUND);
+ return ledgerPendingAcks;
+ }
+
+ private static IntIntPair unpackPendingAckValue(long packedValue) {
+ return IntIntPair.of(unpackRemainingUnacked(packedValue), unpackStickyKeyHash(packedValue));
+ }
+
+ private static int unpackRemainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ private static int unpackStickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
Review Comment:
Good point, thanks. I separated the packed pending-ack value handling from PendingAcksMap by adding a package-private PendingAckValue helper for packing/unpacking, sentinel handling, and pair materialization. I also added PendingAckValueTest to cover round trips, signed sticky-key hashes, sentinel behavior, and rejecting negative remainingUnacked values, while keeping PendingAcksMap focused on the map, locking, and handler logic.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#issuecomment-4701363053
Follow-up plan for context:
This PR intentionally keeps the current nested `TreeMap` structure and only changes the stored pending-ack value representation plus the consumer hot-path accessors. The goal is to make this step reviewable on its own: packed value semantics, signed sticky-key hash handling, zero remaining-count handling, and no behavior change for ledger/entry ordering.
The next steps are planned separately:
1. #26028 adds the local primitive `Long2LongOpenHashMap` utility without changing broker behavior.
2. A follow-up PR will use that primitive map inside `PendingAcksMap` for the per-ledger entry map. That step removes the boxed inner-map key/value overhead and will be based on this packed-value shape once it is stable.
3. Another follow-up PR will add the BitSet prefix index for same-ledger `removeAllUpTo` cleanup. The primitive map improves exact-key operations, but it does not provide the old `TreeMap.headMap(...).clear()` prefix path; the BitSet index is meant to recover that practical cleanup performance while remaining only a performance index.
The O(1) `size()` work is tracked separately in #26019, so this PR does not include any size-accounting changes.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#issuecomment-4857764918
@merlimat I updated the JMH benchmark and reran it on a Linux server with NUMA pinning. The previous numbers were from a local macOS run and may have more noise, so please refer to the latest PR description for the updated benchmark data.
For `forEachScan`, the latest numbers are:
- former layout: `563.022 +/- 41.151 us/op`, `1.932 B/op`
- packed layout: `354.509 +/- 16.877 us/op`, `2.946 B/op`
The previous `815.4 B/op` result does not reproduce with the updated benchmark.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3498932372
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -131,6 +136,7 @@ public interface PendingAcksConsumer {
* @return true if the pending ack was added, and it's allowed to send a message, false otherwise
*/
public boolean addPendingAckIfAllowed(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) {
+ long packedValue = PendingAckValue.pack(remainingUnacked, stickyKeyHash);
Review Comment:
why is the variable introduced a lot earlier than the usage? a typical best practice is to introduce a variable in the scope where it's used and "close" to the usage.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3497384906
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -504,6 +571,28 @@ private void internalRemoveAllUpTo(long markDeleteLedgerId, long markDeleteEntry
}
}
+ private static long packPendingAckValue(int remainingUnacked, int stickyKeyHash) {
+ return ((long) remainingUnacked << Integer.SIZE) | (stickyKeyHash & STICKY_KEY_HASH_MASK);
+ }
+
+ private static Long2LongSortedMap newLedgerPendingAcks() {
+ Long2LongRBTreeMap ledgerPendingAcks = new Long2LongRBTreeMap();
+ ledgerPendingAcks.defaultReturnValue(PACKED_PENDING_ACK_NOT_FOUND);
+ return ledgerPendingAcks;
+ }
+
+ private static IntIntPair unpackPendingAckValue(long packedValue) {
+ return IntIntPair.of(unpackRemainingUnacked(packedValue), unpackStickyKeyHash(packedValue));
+ }
+
+ private static int unpackRemainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ private static int unpackStickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
Review Comment:
The code would be cleaner if a new class would be created that abstracts the packing and unpacking. Unit testing the class is also easier and cleaner. This class could reside as package protected class and if there's a need to reuse it, it could be moved to `pulsar-common` later.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part 2: Pack PendingAcksMap values into long [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3498937492
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAckValue.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+
+final class PendingAckValue {
+ static final int NOT_FOUND = -1;
+ static final long PACKED_NOT_FOUND = packUnchecked(NOT_FOUND, 0);
+
+ private static final long STICKY_KEY_HASH_MASK = 0xFFFF_FFFFL;
+
+ private PendingAckValue() {
+ }
+
+ static long pack(int remainingUnacked, int stickyKeyHash) {
+ if (remainingUnacked < 0) {
+ throw new IllegalArgumentException("remainingUnacked must be non-negative");
+ }
+ return packUnchecked(remainingUnacked, stickyKeyHash);
+ }
+
+ static boolean isNotFound(long packedValue) {
+ return packedValue == PACKED_NOT_FOUND;
+ }
+
+ static IntIntPair toPair(long packedValue) {
+ return IntIntPair.of(remainingUnacked(packedValue), stickyKeyHash(packedValue));
+ }
+
+ static int remainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ static int stickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
+
+ private static long packUnchecked(int remainingUnacked, int stickyKeyHash) {
+ return ((long) remainingUnacked << Integer.SIZE) | (stickyKeyHash & STICKY_KEY_HASH_MASK);
+ }
Review Comment:
document the methods so that future maintainers understand this logic.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -184,15 +190,15 @@ public void forEach(PendingAcksConsumer processor) {
private void processPendingAcks(PendingAcksConsumer processor) {
// this code uses for loops intentionally, don't refactor to use forEach
// iterate the outer map
- for (Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>> entry : pendingAcks.long2ObjectEntrySet()) {
+ for (Long2ObjectMap.Entry<Long2LongSortedMap> entry : pendingAcks.long2ObjectEntrySet()) {
long ledgerId = entry.getLongKey();
- Long2ObjectSortedMap<IntIntPair> ledgerPendingAcks = entry.getValue();
+ Long2LongSortedMap ledgerPendingAcks = entry.getValue();
// iterate the inner map
- for (Long2ObjectMap.Entry<IntIntPair> e : ledgerPendingAcks.long2ObjectEntrySet()) {
+ for (Long2LongMap.Entry e : ledgerPendingAcks.long2LongEntrySet()) {
Review Comment:
use `it.unimi.dsi.fastutil.longs.LongLongBiConsumer` for specific method for iterating `Long2LongMap` entries. Fastutil has some internal optimization for some implementations of `Long2LongMap`. (perhaps it doesn't matter in the end from performance perspective in this case)
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAckValue.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+
+final class PendingAckValue {
Review Comment:
document the purpose. The name of the class `PendingAckValue` isn't optimal since it causes some confusion since the expectation based on the name would be that this is a value object class.
In some naming conventions, classes with only static methods use a plural form. For example `PendingAckValues`. This type of naming would make it harder to misinterprete the purpose and mix this with a value object class.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -98,7 +103,7 @@ public interface PendingAcksConsumer {
}
private final Consumer consumer;
- private final Long2ObjectSortedMap<Long2ObjectSortedMap<IntIntPair>> pendingAcks;
+ private final Long2ObjectSortedMap<Long2LongSortedMap> pendingAcks;
Review Comment:
add a comment above this field to explain the datastructure (what the key is and what the value is). then explain how the Long2LongSortedMap is used to store the `remainingUnacked` and `stickyKeyHash` values for an `entryId` and that the "packing" and "unpacking" operations are handled by the `PendingAckValues` static methods.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3500665171
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAckValues.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+
+/**
+ * Utility methods for storing pending ack values in primitive maps.
+ *
+ * <p>The high 32 bits contain the remaining unacked count and the low 32 bits contain the sticky key hash.
+ * A value with {@link #NOT_FOUND} as the remaining unacked count and 0 as the sticky key hash is reserved
+ * as the missing-entry sentinel.
+ */
+final class PendingAckValues {
+ /**
+ * Remaining unacked count reserved for missing entries.
+ */
+ static final int NOT_FOUND = -1;
+
+ /**
+ * Packed value reserved for missing entries in primitive maps.
+ */
+ static final long PACKED_NOT_FOUND = packUnchecked(NOT_FOUND, 0);
+
+ private static final long STICKY_KEY_HASH_MASK = 0xFFFF_FFFFL;
+
+ private PendingAckValues() {
+ }
+
+ /**
+ * Packs a pending ack's remaining unacked count and sticky key hash into one long value.
+ *
+ * @throws IllegalArgumentException if the remaining unacked count is negative
+ */
+ static long pack(int remainingUnacked, int stickyKeyHash) {
+ if (remainingUnacked < 0) {
+ throw new IllegalArgumentException("remainingUnacked must be non-negative");
+ }
+ return packUnchecked(remainingUnacked, stickyKeyHash);
+ }
+
+ /**
+ * Returns true if the packed value is the missing-entry sentinel.
+ */
+ static boolean isNotFound(long packedValue) {
+ return packedValue == PACKED_NOT_FOUND;
+ }
+
+ /**
+ * Returns the unpacked remaining unacked count and sticky key hash as an {@link IntIntPair}.
+ */
+ static IntIntPair toPair(long packedValue) {
+ return IntIntPair.of(remainingUnacked(packedValue), stickyKeyHash(packedValue));
+ }
+
+ /**
+ * Returns the remaining unacked count stored in the high 32 bits.
+ */
+ static int remainingUnacked(long packedValue) {
+ return (int) (packedValue >> Integer.SIZE);
+ }
+
+ /**
+ * Returns the sticky key hash stored in the low 32 bits.
+ */
+ static int stickyKeyHash(long packedValue) {
+ return (int) packedValue;
+ }
+
+ private static long packUnchecked(int remainingUnacked, int stickyKeyHash) {
Review Comment:
Even better would be to inline this method unless there's another usage.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize PendingAcksMap value storage with primitive longs [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#issuecomment-4845062847
@void-ptr974 Please avoid force pushes to a PR branch when the review of the PR has started. It makes easier for reviewers to track the actual changes since the last time the reviewer performed a review.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #26030:
URL: https://github.com/apache/pulsar/pull/26030
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3508000508
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -182,18 +194,12 @@ public void forEach(PendingAcksConsumer processor) {
// iterate all pending acks and process them
private void processPendingAcks(PendingAcksConsumer processor) {
- // this code uses for loops intentionally, don't refactor to use forEach
- // iterate the outer map
- for (Long2ObjectMap.Entry<Long2ObjectSortedMap<IntIntPair>> entry : pendingAcks.long2ObjectEntrySet()) {
+ for (Long2ObjectMap.Entry<Long2LongSortedMap> entry : pendingAcks.long2ObjectEntrySet()) {
long ledgerId = entry.getLongKey();
- Long2ObjectSortedMap<IntIntPair> ledgerPendingAcks = entry.getValue();
- // iterate the inner map
- for (Long2ObjectMap.Entry<IntIntPair> e : ledgerPendingAcks.long2ObjectEntrySet()) {
- long entryId = e.getLongKey();
- IntIntPair batchSizeAndStickyKeyHash = e.getValue();
- processor.accept(ledgerId, entryId, batchSizeAndStickyKeyHash.leftInt(),
- batchSizeAndStickyKeyHash.rightInt());
- }
+ LongLongBiConsumer pendingAckProcessor = (entryId, packedValue) ->
+ processor.accept(ledgerId, entryId, PendingAckValues.remainingUnacked(packedValue),
+ PendingAckValues.stickyKeyHash(packedValue));
+ entry.getValue().forEach(pendingAckProcessor);
}
Review Comment:
**This `forEach` refactor reintroduces per-ledger lambda allocation on a hot path.**
I initially suggested this change, but it wasn't a good suggestion after all.
This replaced the hand-written nested `for` loops with a `forEach` call and, in the same hunk, removed the `// this code uses for loops intentionally, don't refactor to use forEach` comment that guarded exactly this. The refactor allocates **two lambda objects per ledger, per scan**, where the old nested loops allocated none:
1. **`pendingAckProcessor`** captures `ledgerId` (a per-iteration local) and `processor`, so it's a *capturing* lambda — the JVM creates a fresh instance on every outer-loop iteration; unlike a non-capturing lambda it can't be lowered to a cached constant.
2. **`entry.getValue().forEach(pendingAckProcessor)`** resolves to fastutil's *default* `Long2LongMap.forEach(LongLongBiConsumer)`, because `Long2LongRBTreeMap` doesn't override it. That default allocates a second `wrappingConsumer` lambda per call (`Long2LongMap.java` L268-274), and since the tree map's entry set isn't a `FastEntrySet` it takes the plain `entrySet.forEach(...)` branch — no allocation-free `fastForEach` fast path.
So the cost is `2 × (active ledgers)` net-new lambda allocations per scan, where the old code allocated zero. That's the source of the extra `B/op` on `forEachScan` (`1.932 → 2.946 B/op`, and the earlier `815 B/op` run @merlimat asked about). The benchmark's `ledgers=1` plus a monomorphic call site lets escape analysis scalar-replace most of it, so the residual looks tiny — but the call goes through the `Long2LongSortedMap` interface into a multi-level fastutil call chain, so EA is fragile under real load and the allocation grows with the ledger count.
It matters because `processPendingAcks` backs `forEach` / `forEachAndClear` / `forEachAndClose`, which run on real dispatcher paths — redelivery of unacked messages and consumer disconnect/removal (`PersistentDispatcherMultipleConsumers`, `PersistentStickyKeyDispatcherMultipleConsumers`, `Consumer#redeliverUnacknowledgedMessages`), not just the microbenchmark.
Suggested fix — restore the allocation-free nested loop (this mirrors the pattern already used in `internalRemoveAllUpTo`, which iterates `long2LongEntrySet()` directly):
```suggestion
// this code uses for loops intentionally, don't refactor to use forEach
for (Long2ObjectMap.Entry<Long2LongSortedMap> entry : pendingAcks.long2ObjectEntrySet()) {
long ledgerId = entry.getLongKey();
for (Long2LongMap.Entry e : entry.getValue().long2LongEntrySet()) {
long packedValue = e.getLongValue();
processor.accept(ledgerId, e.getLongKey(),
PendingAckValues.remainingUnacked(packedValue), PendingAckValues.stickyKeyHash(packedValue));
}
}
```
After this, `import it.unimi.dsi.fastutil.longs.LongLongBiConsumer;` becomes unused and should be removed (checkstyle `UnusedImports` will fail otherwise). It'd also be worth re-running `PendingAcksMapPackingBenchmark` with `-p ledgers=20` so the regression — and the fix — are actually visible; `ledgers=1` understates it.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Reduce PendingAcksMap memory by replacing IntIntPair values with packed longs [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26030:
URL: https://github.com/apache/pulsar/pull/26030#discussion_r3508043394
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -271,15 +277,39 @@ public boolean contains(long ledgerId, long entryId) {
* @param ledgerId the ledger ID
* @param entryId the entry ID
* @return the pending ack, or null if not found
+ * @deprecated use {@link #getRemainingUnacked(long, long)} when only the remaining unacked count is needed.
+ * This method materializes an {@link IntIntPair}.
*/
+ @Deprecated
public IntIntPair get(long ledgerId, long entryId) {
Review Comment:
**These deprecated pair-returning methods have no remaining production callers after this PR — consider dropping them instead of shipping deprecated dead code.**
After the migration in this PR, none of the `IntIntPair`-materializing methods are reached from production:
- `PendingAcksMap#get` (this method, L284) — the former callers were the `getPendingAcks().get(...)` lookups in `Consumer` (L845, L852), now on `getRemainingUnacked`.
- `PendingAcksMap#removeAndGet` (L422) — only called by the deprecated `Consumer#removePendingAckAndGet` wrapper (`Consumer.java` L1224), which is itself dead.
- `Consumer#removePendingAckAndGet` (`Consumer.java` L1222) — the former callers in `individualAck` / `applyPendingAckCompletions` now use `removePendingAckAndGetRemainingUnacked`; nothing else calls it.
These are broker-internal classes (`org.apache.pulsar.broker.service`), not client-facing or wire API, so keeping them as `@Deprecated` only adds dead surface with no compatibility upside. I'd remove all three and adapt the remaining non-production users:
- **Tests** — `PendingAcksMapTest` uses `get` (L390, L408, L420, L432) and `removeAndGet` (L305, L319, L332, L395) through the `assertPendingAck(IntIntPair, ...)` helper (L466). Re-point these at the primitive accessors (`getRemainingUnacked` / `removeAndGetRemainingUnacked`), asserting `remainingUnacked` directly and, where the sticky-key hash also matters, unpacking through `PendingAckValues`.
- **Benchmark** — `PendingAcksMapPackingBenchmark` keeps a `getPairHit` scenario (L71-73) and a `get` on the `PendingAcksStore` interface / both store impls solely to measure the pair path. If the production `get` goes away, drop `getPairHit` (or keep the pair variant entirely inside the benchmark's own `FormerPendingAcksMapStore` for the before/after comparison, without depending on a production method).
If you're deliberately keeping these for a planned follow-up, feel free to ignore — but the follow-ups you described (inner-map type swap, BitSet prefix index) don't look like they reintroduce a pair caller.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java:
##########
@@ -381,23 +415,54 @@ public boolean remove(long ledgerId, long entryId) {
* @param ledgerId the ledger ID
* @param entryId the entry ID
* @return the removed entry as an IntIntPair (batchSize, stickyKeyHash), or null if not found
+ * @deprecated use {@link #removeAndGetRemainingUnacked(long, long)} when only the remaining unacked count
+ * is needed. This method materializes an {@link IntIntPair}.
*/
+ @Deprecated
public IntIntPair removeAndGet(long ledgerId, long entryId) {
Review Comment:
Part of the same cleanup as my note on `get` above: after this PR the only caller of `removeAndGet` is the deprecated `Consumer#removePendingAckAndGet` wrapper (`Consumer.java` L1224), which itself has no callers — so the whole chain is dead in production. It could be removed together with `get`, adapting `PendingAcksMapTest` (L305, L319, L332, L395) to `removeAndGetRemainingUnacked`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -1192,14 +1193,32 @@ public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelt
return true;
}
+ /**
+ * Atomically remove the pending ack entry and return its stored values.
+ *
+ * <p>No-op if {@code pendingAcks} is not initialized.
+ *
+ * @return the remaining unacked count, or {@link PendingAcksMap#PENDING_ACK_NOT_FOUND} if not found
+ */
+ public int removePendingAckAndGetRemainingUnacked(long ledgerId, long entryId) {
+ if (pendingAcks != null) {
+ return pendingAcks.removeAndGetRemainingUnacked(ledgerId, entryId);
+ }
+ return PENDING_ACK_NOT_FOUND;
+ }
+
/**
* Atomically remove the pending ack entry and return its stored values.
*
* <p>No-op if {@code pendingAcks} is not initialized.
*
* @return the removed {@link IntIntPair#leftInt() remainingUnacked} and
* {@link IntIntPair#rightInt() stickyKeyHash}, or {@code null} if not found
+ * @deprecated use {@link #removePendingAckAndGetRemainingUnacked(long, long)} when only the remaining unacked
+ * count is needed. This method materializes an {@link IntIntPair}.
*/
+ @Deprecated
+ @SuppressWarnings("deprecation")
public IntIntPair removePendingAckAndGet(long ledgerId, long entryId) {
Review Comment:
Same cleanup: after this PR this deprecated wrapper has no callers (the `individualAck` / `applyPendingAckCompletions` sites now use `removePendingAckAndGetRemainingUnacked`), and it only delegates to the likewise-dead `PendingAcksMap#removeAndGet`. Since `Consumer` here is broker-internal, I'd remove it rather than carry it as `@Deprecated` dead code.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org