[
https://issues.apache.org/jira/browse/QPID-8538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17360343#comment-17360343
]
Peter Moran commented on QPID-8538:
-----------------------------------
Citation from AMQP:
" If the sender's drain flag is set and there are no available messages, the
sender MUST advance its delivery-count until link-credit is zero, and send its
updated flow state to the receiver.
(1) If a message is available within the timeout, it will
arrive at this point.
(2) If a message is not available within the timeout, the
drain flag will ensure that the sender promptly advances the
delivery-count until link-credit is consumed.
"
The AMQP documentation is bit unclear how receiver should replenish the credit
or should understand the delivery_count. I understand that subsequent get
should increase the credit if there are messages delivered. But what about the
case above, message were not delivered (just the flow message from broker with
updated deliveries +credit) and credit is depleted and queue is e.g 1 message.
There is just a gap in the fetch which ignores the possibility that broker
cannot deliver messages in given time and just updates the delivery count. I
would any way expect the fetch must therefore always replenish the credit
consider amount of returned messages (obtained by get) and then replenish the
credit immediately, if it does not the the link credit will stay very low and
never returns to capacity. I have tested the code above and I will retest one
more alternative
remove the pn link flow after drain and move it after get ( if (lnk->capacity
&& pn_link_queued(lnk->receiver) == 0) {)
{code:java}
auto res = get(ssn, lnk, message, qpid::messaging::Duration::IMMEDIATE);
{
if (lnk->capacity)
{
sys::Monitor::ScopedLock l(lock);
pn_link_flow(lnk->receiver, lnk->capacity -
pn_link_queued(lnk->receiver));
wakeupDriver();
}
}
return res;
{code}
Alternative consequence if you believe the qpid-cpp is correct then there is a
bug in proton:
{code:java}
if (delta > 0) {
link->state.delivery_count += delta;
link->state.link_credit -= delta;
link->credit -= delta;
link->drained += delta;
{code}
if there are 0 message available then broker delivers and sets delivery count
to maximum possible (current + granted credit) this code in pn_do_flow decrease
the credit by delta but also sets drained to + delta, which does not match any
documentation expectation, from the documentation of pn_link_drained:
_When invoked on a receiving link, this operation will return and reset the
number of credits the sender has released back to the receiver._
which is just returns link->drained, but this is not traded as execs in
pn_do_flow,
Is the documentation of proton wrong? if it is not excess and it is amount of
messages drained from the queue it cannot be calculated from delivery count.
This code does not make any sense to me. delivery count is not counter as you
said yet proton treats it like it is. And consequence of all this is that the
receiver must top up its credit to (capacity - link->queued) after it drains
regardless if the receiver read messages and incremented credit by itself and
then numbers should match, or got nothing because the credit was consumed by
sender because there are no messages.
> Receiver fetch method depletes credit and slows down receiver
> -------------------------------------------------------------
>
> Key: QPID-8538
> URL: https://issues.apache.org/jira/browse/QPID-8538
> Project: Qpid
> Issue Type: Bug
> Components: C++ Client
> Affects Versions: qpid-cpp-1.39.0
> Reporter: Peter Moran
> Priority: Major
>
> As remote broker java broker is used.
> e.g
> capacity of link is 10
> credit is 10
> When calling Method ConnectionContext::fetch and there are no data available
> inside receiver (get returns false) it
> # Enables drain mode pn_link_drain(lnk->receiver, 0);
> # Tries to process all queued messages (while
> (pn_link_draining(lnk->receiver) && !pn_link_queued(lnk->receiver)))
> # Credit is being drained by broker pn_do_flow in proton ( pn_sequence_t
> delta = delivery_count - link->state.delivery_count;) -> delta is 10 and
> credit is cleared to 0, yet no messages arrived from broker at all
> if there are queued messages and credit is 0 then pn_link_draining returns
> false and while stops
> # however if (lnk->capacity && pn_link_queued(lnk->receiver) == 0) replenish
> credit only if there are no outstanding queued messages to the maximum
> capacity, this should be rather capacity - credit, but also the condition
> seems to be wrong, it should not check the queued messages.
> the code should look like this
> {code:java}
> if (lnk->capacity) {
> pn_link_flow(lnk->receiver, lnk->capacity -
> pn_link_credit(lnk->receiver));
> }
> or
> if (lnk->capacity) {
> pn_link_flow(lnk->receiver, pn_link_drained(lnk->receiver));
> }{code}
> this would much more look like what is inside proton c++ api
> {code:java}
> void credit_topup(pn_link_t *link) {
> assert(pn_link_is_receiver(link));
> int window = link_context::get(link).credit_window;
> if (window) {
> int delta = window - pn_link_credit(link);
> pn_link_flow(link, delta);
> }
> }
> // receiver
> if (!pn_link_credit(lnk) && lctx.draining) {
> lctx.draining = false;
> pn_link_set_drain(lnk, false);
> receiver r(make_wrapper<receiver>(lnk));
> handler.on_receiver_drain_finish(r);
> }
> credit_topup(lnk);
> {code}
> there is no condition about queued and also the resulting credit is set to
> the link capacity
> (e.g 10 - delta) delta is delivery count reduced from by pn_do_flow,
> However this to me looks like some incompatibility between client and broker,
> as one would not expect that delta is 10 if no messages are received from
> broker. is this some kind of message sequencing problem?
>
> fyi pn_link_queued messages does not mean they they are locally available
> they might be still on broker and also getAvaialable implementation is wrong.
>
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
