You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "thetumbled (via GitHub)" <gi...@apache.org> on 2024/11/26 08:44:25 UTC

[PR] [improve][broker] Reduce memory occupation of InMemoryRedeliveryTracker. [pulsar]

thetumbled opened a new pull request, #23640:
URL: https://github.com/apache/pulsar/pull/23640

   
   ### Motivation
   
   Reduce memory occupation of the `InMemoryRedeliveryTracker`.
   
   ### Modifications
   
   Use `Long2ObjectMap<Long2IntMap>` to replace `ConcurrentLongLongPairHashMap`.
   
   ### Verifying this change
   
   - [ ] Make sure that the change passes the CI checks.
   
   *(Please pick either of the following options)*
   
   This change is a trivial rework / code cleanup without any test coverage.
   
   ### Does this pull request potentially affect one of the following parts:
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   *If the box was checked, please highlight the changes*
   
   - [ ] Dependencies (add or upgrade a dependency)
   - [ ] The public API
   - [ ] The schema
   - [ ] The default values of configurations
   - [ ] The threading model
   - [ ] The binary protocol
   - [ ] The REST endpoints
   - [ ] The admin CLI options
   - [ ] The metrics
   - [ ] Anything that affects deployment
   
   ### Documentation
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [ ] `doc` <!-- Your PR contains doc changes. -->
   - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
   - [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [ ] `doc-complete` <!-- Docs have been already added -->
   
   ### Matching PR in forked repository
   
   PR in forked repository: 
   
   
   


-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1859696059


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {
+    // ledgerId -> entryId -> count
+    private Long2ObjectMap<Long2IntMap> trackerCache = new Long2ObjectOpenHashMap<>();
 
     @Override
     public int incrementAndGetRedeliveryCount(Position position) {
-        Position positionImpl = position;
-        LongPair count = trackerCache.get(positionImpl.getLedgerId(), positionImpl.getEntryId());
-        int newCount = (int) (count != null ? count.first + 1 : 1);
-        trackerCache.put(positionImpl.getLedgerId(), positionImpl.getEntryId(), newCount, 0L);
+        long stamp = writeLock();
+        int newCount;
+        try {
+            Long2IntMap entryMap = trackerCache.computeIfAbsent(position.getLedgerId(),
+                    k -> new Long2IntOpenHashMap());
+            newCount = entryMap.getOrDefault(position.getEntryId(), 0) + 1;
+            entryMap.put(position.getEntryId(), newCount);
+        } finally {
+            unlockWrite(stamp);
+        }
         return newCount;
     }
 
     @Override
     public int getRedeliveryCount(long ledgerId, long entryId) {
-        LongPair count = trackerCache.get(ledgerId, entryId);
-        return (int) (count != null ? count.first : 0);
+        long stamp = tryOptimisticRead();
+        Long2IntMap entryMap = trackerCache.get(ledgerId);
+        int count = entryMap != null ? entryMap.get(entryId) : 0;

Review Comment:
   All classes from fastutil is not thread safe, not supporting concurrent modification, but support concurrent read. so we can use optimistic read on it.
   https://fastutil.di.unimi.it/docs/#warnings-heading
   `
   All classes are not synchronized. If multiple threads access one of these classes concurrently, and at least one of the threads modifies it, it must be synchronized externally. Iterators will behave unpredictably in the presence of concurrent modifications. Reads, however, can be carried out concurrently.
   `
   



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {
+    // ledgerId -> entryId -> count
+    private Long2ObjectMap<Long2IntMap> trackerCache = new Long2ObjectOpenHashMap<>();
 
     @Override
     public int incrementAndGetRedeliveryCount(Position position) {
-        Position positionImpl = position;
-        LongPair count = trackerCache.get(positionImpl.getLedgerId(), positionImpl.getEntryId());
-        int newCount = (int) (count != null ? count.first + 1 : 1);
-        trackerCache.put(positionImpl.getLedgerId(), positionImpl.getEntryId(), newCount, 0L);
+        long stamp = writeLock();
+        int newCount;
+        try {
+            Long2IntMap entryMap = trackerCache.computeIfAbsent(position.getLedgerId(),
+                    k -> new Long2IntOpenHashMap());
+            newCount = entryMap.getOrDefault(position.getEntryId(), 0) + 1;
+            entryMap.put(position.getEntryId(), newCount);
+        } finally {
+            unlockWrite(stamp);
+        }
         return newCount;
     }
 
     @Override
     public int getRedeliveryCount(long ledgerId, long entryId) {
-        LongPair count = trackerCache.get(ledgerId, entryId);
-        return (int) (count != null ? count.first : 0);
+        long stamp = tryOptimisticRead();
+        Long2IntMap entryMap = trackerCache.get(ledgerId);
+        int count = entryMap != null ? entryMap.get(entryId) : 0;

Review Comment:
   All classes from fastutil is not thread safe, not supporting concurrent modification, but support concurrent read. so we can use optimistic read on it.
   https://fastutil.di.unimi.it/docs/#warnings-heading
   `
   All classes are not synchronized. If multiple threads access one of these classes concurrently, and at least one of the threads modifies it, it must be synchronized externally. Iterators will behave unpredictably in the presence of concurrent modifications. **Reads, however, can be carried out concurrently**.
   `
   



-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1859696059


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {
+    // ledgerId -> entryId -> count
+    private Long2ObjectMap<Long2IntMap> trackerCache = new Long2ObjectOpenHashMap<>();
 
     @Override
     public int incrementAndGetRedeliveryCount(Position position) {
-        Position positionImpl = position;
-        LongPair count = trackerCache.get(positionImpl.getLedgerId(), positionImpl.getEntryId());
-        int newCount = (int) (count != null ? count.first + 1 : 1);
-        trackerCache.put(positionImpl.getLedgerId(), positionImpl.getEntryId(), newCount, 0L);
+        long stamp = writeLock();
+        int newCount;
+        try {
+            Long2IntMap entryMap = trackerCache.computeIfAbsent(position.getLedgerId(),
+                    k -> new Long2IntOpenHashMap());
+            newCount = entryMap.getOrDefault(position.getEntryId(), 0) + 1;
+            entryMap.put(position.getEntryId(), newCount);
+        } finally {
+            unlockWrite(stamp);
+        }
         return newCount;
     }
 
     @Override
     public int getRedeliveryCount(long ledgerId, long entryId) {
-        LongPair count = trackerCache.get(ledgerId, entryId);
-        return (int) (count != null ? count.first : 0);
+        long stamp = tryOptimisticRead();
+        Long2IntMap entryMap = trackerCache.get(ledgerId);
+        int count = entryMap != null ? entryMap.get(entryId) : 0;

Review Comment:
   All classes from fastutil is not thread safe, not supporting concurrent modification, but support concurrent read. so we can use optimistic read on it.
   https://fastutil.di.unimi.it/docs/#warnings-heading
   `
   All classes are not synchronized. If multiple threads access one of these classes concurrently, and at least one of the threads modifies it, it must be synchronized externally. Iterators will behave unpredictably in the presence of concurrent modifications. Reads, however, can be carried out concurrently.
   `
   



-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #23640:
URL: https://github.com/apache/pulsar/pull/23640


-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#issuecomment-2503532330

   ## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/23640?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   Attention: Patch coverage is `63.15789%` with `14 lines` in your changes missing coverage. Please review.
   > Project coverage is 74.34%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`c37e605`)](https://app.codecov.io/gh/apache/pulsar/commit/c37e6050d9a9f7a7becbe845bf1024170312218d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   > Report is 750 commits behind head on master.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/23640?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...lsar/broker/service/InMemoryRedeliveryTracker.java](https://app.codecov.io/gh/apache/pulsar/pull/23640?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FInMemoryRedeliveryTracker.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0luTWVtb3J5UmVkZWxpdmVyeVRyYWNrZXIuamF2YQ==) | 63.15% | [12 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/23640?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   
   <details><summary>Additional details and impacted files</summary>
   
   
   [![Impacted file tree graph](https://app.codecov.io/gh/apache/pulsar/pull/23640/graphs/tree.svg?width=650&height=150&src=pr&token=acYqCpsK9J&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)](https://app.codecov.io/gh/apache/pulsar/pull/23640?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #23640      +/-   ##
   ============================================
   + Coverage     73.57%   74.34%   +0.76%     
   - Complexity    32624    34997    +2373     
   ============================================
     Files          1877     1944      +67     
     Lines        139502   147224    +7722     
     Branches      15299    16246     +947     
   ============================================
   + Hits         102638   109453    +6815     
   - Misses        28908    29322     +414     
   - Partials       7956     8449     +493     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pulsar/pull/23640/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [inttests](https://app.codecov.io/gh/apache/pulsar/pull/23640/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `27.22% <34.21%> (+2.63%)` | :arrow_up: |
   | [systests](https://app.codecov.io/gh/apache/pulsar/pull/23640/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.29% <63.15%> (-0.03%)` | :arrow_down: |
   | [unittests](https://app.codecov.io/gh/apache/pulsar/pull/23640/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.73% <63.15%> (+0.88%)` | :arrow_up: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/23640?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [...lsar/broker/service/InMemoryRedeliveryTracker.java](https://app.codecov.io/gh/apache/pulsar/pull/23640?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FInMemoryRedeliveryTracker.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0luTWVtb3J5UmVkZWxpdmVyeVRyYWNrZXIuamF2YQ==) | `65.11% <63.15%> (-30.13%)` | :arrow_down: |
   
   ... and [660 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/23640/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   </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] Reduce memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1860287607


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,95 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
 public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)

Review Comment:
   It seems that `Long2ObjectOpenHashMap` support auto shrink.
   `If the table is emptied below one fourth of the load factor, it is halved in size` from the class document: 
   `
   The table is filled up to a specified load factor, and then doubled in size to accommodate new entries. If the table is emptied below one fourth of the load factor, it is halved in size; however, the table is never reduced to a size smaller than that at creation time: this approach makes it possible to create maps with a large capacity in which insertions and deletions do not cause immediately rehashing. Moreover, halving is not performed when deleting entries from an iterator, as it would interfere with the iteration process.
   `



-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1860216615


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,95 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
 public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)

Review Comment:
   `Long2ObjectOpenHashMap` contains `trim()` and `trim(int)` methods to rehash the table if it's too large. I wonder if it would be useful to implement a feature similar to the `ConcurrentLongLongPairHashMap`'s `autoShrink` feature using the `trim(int)` method?



-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1859696059


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {
+    // ledgerId -> entryId -> count
+    private Long2ObjectMap<Long2IntMap> trackerCache = new Long2ObjectOpenHashMap<>();
 
     @Override
     public int incrementAndGetRedeliveryCount(Position position) {
-        Position positionImpl = position;
-        LongPair count = trackerCache.get(positionImpl.getLedgerId(), positionImpl.getEntryId());
-        int newCount = (int) (count != null ? count.first + 1 : 1);
-        trackerCache.put(positionImpl.getLedgerId(), positionImpl.getEntryId(), newCount, 0L);
+        long stamp = writeLock();
+        int newCount;
+        try {
+            Long2IntMap entryMap = trackerCache.computeIfAbsent(position.getLedgerId(),
+                    k -> new Long2IntOpenHashMap());
+            newCount = entryMap.getOrDefault(position.getEntryId(), 0) + 1;
+            entryMap.put(position.getEntryId(), newCount);
+        } finally {
+            unlockWrite(stamp);
+        }
         return newCount;
     }
 
     @Override
     public int getRedeliveryCount(long ledgerId, long entryId) {
-        LongPair count = trackerCache.get(ledgerId, entryId);
-        return (int) (count != null ? count.first : 0);
+        long stamp = tryOptimisticRead();
+        Long2IntMap entryMap = trackerCache.get(ledgerId);
+        int count = entryMap != null ? entryMap.get(entryId) : 0;

Review Comment:
   All classes from fastutil is not thread safe, not supporting concurrent modification, but support concurrent read. so we can use optimistic read on it.
   https://fastutil.di.unimi.it/docs/#warnings-heading
   ```
   All classes are not synchronized. If multiple threads access one of these classes concurrently, and at least one of the threads modifies it, it must be synchronized externally. Iterators will behave unpredictably in the presence of concurrent modifications. Reads, however, can be carried out concurrently.
   ```
   



-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1859693275


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {

Review Comment:
   good idea, updated.



-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1858986420


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {
+    // ledgerId -> entryId -> count
+    private Long2ObjectMap<Long2IntMap> trackerCache = new Long2ObjectOpenHashMap<>();
 
     @Override
     public int incrementAndGetRedeliveryCount(Position position) {
-        Position positionImpl = position;
-        LongPair count = trackerCache.get(positionImpl.getLedgerId(), positionImpl.getEntryId());
-        int newCount = (int) (count != null ? count.first + 1 : 1);
-        trackerCache.put(positionImpl.getLedgerId(), positionImpl.getEntryId(), newCount, 0L);
+        long stamp = writeLock();
+        int newCount;
+        try {
+            Long2IntMap entryMap = trackerCache.computeIfAbsent(position.getLedgerId(),
+                    k -> new Long2IntOpenHashMap());
+            newCount = entryMap.getOrDefault(position.getEntryId(), 0) + 1;
+            entryMap.put(position.getEntryId(), newCount);
+        } finally {
+            unlockWrite(stamp);
+        }
         return newCount;
     }
 
     @Override
     public int getRedeliveryCount(long ledgerId, long entryId) {
-        LongPair count = trackerCache.get(ledgerId, entryId);
-        return (int) (count != null ? count.first : 0);
+        long stamp = tryOptimisticRead();
+        Long2IntMap entryMap = trackerCache.get(ledgerId);
+        int count = entryMap != null ? entryMap.get(entryId) : 0;

Review Comment:
   Some data structures might not support an optimistic read. The concern is that a data race could lead to a infinite loop. For example, this used to be the case with JDK's HashMap very long time ago. It might be worth validating this with a concurrency test or by checking the documentation.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {

Review Comment:
   It's always better to use composition instead of inheritance, especially in this case. I don't think it's a good idea to extend `StampedLock`.



-- 
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 memory occupation of InMemoryRedeliveryTracker. [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23640:
URL: https://github.com/apache/pulsar/pull/23640#discussion_r1860209903


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/InMemoryRedeliveryTracker.java:
##########
@@ -18,49 +18,94 @@
  */
 package org.apache.pulsar.broker.service;
 
+import it.unimi.dsi.fastutil.longs.Long2IntMap;
+import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import java.util.List;
+import java.util.concurrent.locks.StampedLock;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 
-public class InMemoryRedeliveryTracker implements RedeliveryTracker {
-
-    private ConcurrentLongLongPairHashMap trackerCache = ConcurrentLongLongPairHashMap.newBuilder()
-            .concurrencyLevel(1)
-            .expectedItems(256)
-            .autoShrink(true)
-            .build();
+public class InMemoryRedeliveryTracker extends StampedLock implements RedeliveryTracker {
+    // ledgerId -> entryId -> count
+    private Long2ObjectMap<Long2IntMap> trackerCache = new Long2ObjectOpenHashMap<>();
 
     @Override
     public int incrementAndGetRedeliveryCount(Position position) {
-        Position positionImpl = position;
-        LongPair count = trackerCache.get(positionImpl.getLedgerId(), positionImpl.getEntryId());
-        int newCount = (int) (count != null ? count.first + 1 : 1);
-        trackerCache.put(positionImpl.getLedgerId(), positionImpl.getEntryId(), newCount, 0L);
+        long stamp = writeLock();
+        int newCount;
+        try {
+            Long2IntMap entryMap = trackerCache.computeIfAbsent(position.getLedgerId(),
+                    k -> new Long2IntOpenHashMap());
+            newCount = entryMap.getOrDefault(position.getEntryId(), 0) + 1;
+            entryMap.put(position.getEntryId(), newCount);
+        } finally {
+            unlockWrite(stamp);
+        }
         return newCount;
     }
 
     @Override
     public int getRedeliveryCount(long ledgerId, long entryId) {
-        LongPair count = trackerCache.get(ledgerId, entryId);
-        return (int) (count != null ? count.first : 0);
+        long stamp = tryOptimisticRead();
+        Long2IntMap entryMap = trackerCache.get(ledgerId);
+        int count = entryMap != null ? entryMap.get(entryId) : 0;

Review Comment:
   Thanks for checking, everything fine here.



-- 
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