You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@activemq.apache.org by wy96f <gi...@git.apache.org> on 2018/10/12 12:34:53 UTC
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
GitHub user wy96f opened a pull request:
https://github.com/apache/activemq-artemis/pull/2369
ARTEMIS-2123 Paging not stopped if there are no consumers on one subscription
Reproduction steps:
1. create a topic t and two subscriptions ta, tb
2. send messages to both ta and tb
3. create ta consumers and receive all messages from ta
4. close consumers
5. only send message to tb
6. create tb consumers and receive all message from tb
7. topic not stopped paging
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/wy96f/activemq-artemis not_stop_paging_without_consumers_on_subscription
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/activemq-artemis/pull/2369.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2369
----
commit 6aa65c969073c2908520156ce5113a7c63a71151
Author: yang wei <wy...@...>
Date: 2018-10-12T12:32:02Z
ARTEMIS-2123 Paging not stopped if there are no consumers on one subscription
----
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/activemq-artemis/pull/2369
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by wy96f <gi...@git.apache.org>.
Github user wy96f commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224977450
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
--- End diff --
> this is saving a deliveryAsync call.. but I"m just wondering if it would be too costly to do the check on every queue?
All PageSubcriptions use the same executor and it will do many jobs such as PageSubscriptionImpl::cleanupEntries, DepageRunner, DeliverRunner, ExpiryScanner, PageCursorProviderImpl::scheduleCleanup, PageCursorProviderImpl::cleanup, etc. We'd better prevent these extra needless deliverAsync calls.
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by wy96f <gi...@git.apache.org>.
Github user wy96f commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r225769985
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,18 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList, long minPage) {
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+ if (firstPage == minPage) {
+ if (cursor.getQueue().getMessageCount() == 0 && cursorList.size() > 1) {
+ cursor.getQueue().deliverAsync();
+ break;
--- End diff --
@wy96f @clebertsuconic Maybe more than one. I changed the code and added another test. Please review the code:)
---
[GitHub] activemq-artemis issue #2369: ARTEMIS-2123 Paging not stopped if there are n...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on the issue:
https://github.com/apache/activemq-artemis/pull/2369
@wy96f are you waiting a release to apply your patch?
---
[GitHub] activemq-artemis issue #2369: ARTEMIS-2123 Paging not stopped if there are n...
Posted by wy96f <gi...@git.apache.org>.
Github user wy96f commented on the issue:
https://github.com/apache/activemq-artemis/pull/2369
@clebertsuconic @franz1981 Hi, this bug occurred in our production environment yesterday. Our topic has one subscription without messages at the moment. Though the consumers subscribed to it and all messages are consumed, paging is not stopped causing more than 6000 pages and 200000000 large messages related to pages are not deleted.
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224828919
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
+ slowSubscription.getQueue().deliverAsync();
--- End diff --
@wy96f how do I get to meet you? :) IRC / talk... etc? you have been providing great fixes.
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224839730
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
--- End diff --
@clebertsuconic @wy96f
Not sure but deliverIfNecessary could be written simlar to this one?
```
private void deliverIfNecessary(Collection<PageSubscription> cursorList, long minPage) {
for (PageSubscription cursor : cursorList) {
long firstPage = cursor.getFirstPage();
if (firstPage == minPage) {
if (cursor.getQueue().getMessageCount() == 0) {
cursor.getQueue().deliverAsync();
break;
}
}
}
}
```
As long as minPage is not Long.MAX_VALUE (it shouldn't be)...
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by wy96f <gi...@git.apache.org>.
Github user wy96f commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224976264
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
+ slowSubscription.getQueue().deliverAsync();
--- End diff --
> Just to understand the fix: given that QueueImpl::deliverAsync will trigger QueueImpl::checkDepage that will schedule an async task QueueImpl::DepageRunner how do you know that it will be finished time in order to have the PageSubscription completed when it will check later on the same cleanup call?
> From what I have understood it will be eventually completed on the cleanup that will follow a successfull DepageRunner::run...
Yes, that's right. A successful DepageRunner::run will trigger PageSubscription::processACK, then PageSubscription::cleanupEntries, and lastly PageCursorProvider::cleanup.
---
[GitHub] activemq-artemis issue #2369: ARTEMIS-2123 Paging not stopped if there are n...
Posted by wy96f <gi...@git.apache.org>.
Github user wy96f commented on the issue:
https://github.com/apache/activemq-artemis/pull/2369
> @wy96f are you waiting a release to apply your patch?
@clebertsuconic We have an internal branch. We will upgrade our cluster after a month.
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224827924
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
+ slowSubscription.getQueue().deliverAsync();
--- End diff --
@wy96f man, you're good!
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224832278
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
--- End diff --
this is saving a deliveryAsync call.. but I"m just wondering if it would be too costly to do the check on every queue?
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r225077165
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
--- End diff --
@wy96f nice! So please change the code with the refined version of `deliverIfNecessary` and squash the commits into one :+1:
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by wy96f <gi...@git.apache.org>.
Github user wy96f commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224976508
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
+ slowSubscription.getQueue().deliverAsync();
--- End diff --
@clebertsuconic Thank you man. We're heavily using artemis in our company. It's a great product and thanks for your work. We are a team: shoukunhuai, me, shoukunhuai. We will keep on developing the code :)
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r225419711
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,18 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList, long minPage) {
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+ if (firstPage == minPage) {
+ if (cursor.getQueue().getMessageCount() == 0 && cursorList.size() > 1) {
+ cursor.getQueue().deliverAsync();
+ break;
--- End diff --
The `break` could be taken out, landing into `if (firstPage == minPage) { ..break; }`?
@wy96f @clebertsuconic how many cursors can have `cursor.getFirstPage() == minPage`?
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224825736
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
+ slowSubscription.getQueue().deliverAsync();
--- End diff --
Just to understand the fix: given that QueueImpl::deliverAsync will trigger QueueImpl::checkDepage that will schedule an async task QueueImpl::DepageRunner how do you know that it will be finished time in order to have the PageSubscription completed?
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224830396
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
+ slowSubscription.getQueue().deliverAsync();
--- End diff --
@wy96f @clebertsuconic totally agree!
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224832160
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
--- End diff --
I was talking to @franz1981 here where I am.. couldn't we just call deliverAsync on every queue after cleanup?
---
[GitHub] activemq-artemis pull request #2369: ARTEMIS-2123 Paging not stopped if ther...
Posted by wy96f <gi...@git.apache.org>.
Github user wy96f commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/2369#discussion_r224977706
--- Diff: artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java ---
@@ -599,6 +600,29 @@ private long checkMinPage(Collection<PageSubscription> cursorList) {
}
+ private void deliverIfNecessary(Collection<PageSubscription> cursorList) {
+ long minPage = Long.MAX_VALUE;
+ PageSubscription slowSubscription = null;
+ int nonEmptyCursorNum = 0;
+
+ for (PageSubscription cursor : cursorList) {
+ long firstPage = cursor.getFirstPage();
+
+ // the cursor will return -1 if the cursor is empty
+ if (firstPage >= 0) {
+ nonEmptyCursorNum++;
+ if (firstPage < minPage) {
+ minPage = firstPage;
+ slowSubscription = cursor.getQueue().getMessageCount() == 0 ? cursor : null;
+ }
+ }
+ }
+
+ if (nonEmptyCursorNum > 1 && slowSubscription != null) {
--- End diff --
> @clebertsuconic @wy96f
> Not sure but deliverIfNecessary could be written simlar to this one?
>
> ```
> private void deliverIfNecessary(Collection<PageSubscription> cursorList, long minPage) {
> for (PageSubscription cursor : cursorList) {
> long firstPage = cursor.getFirstPage();
> if (firstPage == minPage) {
> if (cursor.getQueue().getMessageCount() == 0) {
> cursor.getQueue().deliverAsync();
> break;
> }
> }
> }
> }
> ```
> As long as minPage is not Long.MAX_VALUE (it shouldn't be)...
@clebertsuconic @franz1981 Great, it's more precise. We can also judge whether cursorList.size() > 1 bcs for the queue or topic with only one subscription in which case cursorList.size() == 1 we don't need to call deliverAsync again.
---