You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gm...@apache.org on 2018/09/10 13:49:37 UTC
qpid-dispatch git commit: DISPATCH-1090 - Drain excess credit if
there are no receivers. This closes #371.
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 06bfb5414 -> 0a8d43285
DISPATCH-1090 - Drain excess credit if there are no receivers. This closes #371.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/0a8d4328
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/0a8d4328
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/0a8d4328
Branch: refs/heads/master
Commit: 0a8d4328580f805d490be32d7b8af7cbcc0e4d64
Parents: 06bfb54
Author: Ganesh Murthy <gm...@redhat.com>
Authored: Tue Aug 28 15:09:16 2018 -0400
Committer: Ganesh Murthy <gm...@redhat.com>
Committed: Mon Sep 10 09:48:55 2018 -0400
----------------------------------------------------------------------
src/router_core/transfer.c | 8 +++-
tests/system_tests_drain.py | 94 +++++++++++++++++++++++++++++++++++++++-
2 files changed, 99 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/0a8d4328/src/router_core/transfer.c
----------------------------------------------------------------------
diff --git a/src/router_core/transfer.c b/src/router_core/transfer.c
index 92b320c..2031b0d 100644
--- a/src/router_core/transfer.c
+++ b/src/router_core/transfer.c
@@ -835,8 +835,13 @@ static void qdr_link_forward_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery
} else {
if (more)
dlv->disposition = PN_RELEASED;
- else
+ else {
qdr_delivery_release_CT(core, dlv);
+ //
+ // Drain credit on the link
+ //
+ qdr_link_issue_credit_CT(core, link, 0, true);
+ }
}
if (qdr_is_addr_treatment_multicast(link->owning_addr))
@@ -949,7 +954,6 @@ static void qdr_link_forward_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery
//
// Again, don't bother decrementing then incrementing the ref_count
//
-
DEQ_INSERT_TAIL(link->unsettled, dlv);
dlv->where = QDR_DELIVERY_IN_UNSETTLED;
qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer: dlv:%lx qdr_link_forward_CT: action-list -> unsettled-list", (long) dlv);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/0a8d4328/tests/system_tests_drain.py
----------------------------------------------------------------------
diff --git a/tests/system_tests_drain.py b/tests/system_tests_drain.py
index 1b38aba..bdc56ed 100644
--- a/tests/system_tests_drain.py
+++ b/tests/system_tests_drain.py
@@ -27,7 +27,9 @@ from system_test import TestCase, Qdrouterd, main_module
from system_tests_drain_support import DrainMessagesHandler, DrainOneMessageHandler
from system_tests_drain_support import DrainNoMessagesHandler, DrainNoMoreMessagesHandler
from system_tests_drain_support import DrainMessagesMoreHandler
-
+from proton.handlers import MessagingHandler
+from proton import Message
+from proton.reactor import Container
from time import sleep
class DrainSupportTest(TestCase):
@@ -109,5 +111,95 @@ class DrainSupportTest(TestCase):
self.assertEqual(drain_support.error, None)
+class ReceiverDropsOffDrainTest(TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ super(ReceiverDropsOffDrainTest, cls).setUpClass()
+ config = Qdrouterd.Config([
+ ('router', {'mode': 'standalone', 'id': 'Broker'}),
+ ('listener', {'role': 'normal',
+ 'port': cls.tester.get_port(),
+ 'saslMechanisms': 'ANONYMOUS'}),
+ ])
+
+ cls.router = cls.tester.qdrouterd("A", config, wait=True)
+ cls.address = cls.router.addresses[0]
+
+ def test_receiver_drops_off_sender_receives_drain(self):
+ test = ReceiverDropsOffSenderDrain(self.address, "examples")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+class ReceiverDropsOffSenderDrain(MessagingHandler):
+ def __init__(self, address, dest):
+ super(ReceiverDropsOffSenderDrain, self).__init__()
+ self.sender_conn = None
+ self.receiver_conn = None
+ self.sender = None
+ self.receiver = None
+ self.error = None
+ self.sender_drained = False
+ self.address = address
+ self.dest = dest
+ self.num_msgs = 0
+ self.receiver_closed = False
+ self.drained = 0
+ self.expected_drained = 249
+
+ def timeout(self):
+ if not self.error:
+ self.error = "Timeout Expired: Sender was not drained. Expected " \
+ "drained=%s, actual drain=%s " % \
+ (self.expected_drained, self.drained)
+ self.sender_conn.close()
+ self.receiver_conn.close()
+
+ def on_start(self, event):
+ # Create sender and receiver in two separate connections
+ self.sender_conn = event.container.connect(self.address)
+ self.receiver_conn = event.container.connect(self.address)
+ self.sender = event.container.create_sender(self.sender_conn,
+ self.dest)
+ self.receiver = event.container.create_receiver(self.receiver_conn,
+ self.dest)
+
+ def on_sendable(self, event):
+ # Send just one message for now
+ if self.num_msgs < 1:
+ self.num_msgs += 1
+ msg = Message(body={'number': 1})
+ self.sender.send(msg)
+
+ def on_message(self, event):
+ # As soon as the receiver receives the message, close the receiver
+ if event.receiver == self.receiver:
+ self.receiver.close()
+ self.receiver_closed = True
+
+ def on_link_closed(self, event):
+ if event.receiver == self.receiver:
+ # The receiver link is closed. The sender still have a credit
+ # of 249. Now send a message. The receiver is already gone at
+ # this point. The router will receive this message and see that
+ # there are no receivers and it will send a drain to the sender
+ # This test will not work without the fix for DISPATCH-1090
+ msg = Message(body={'number': 2})
+ self.sender.send(msg)
+
+ def on_link_flow(self, event):
+ if self.receiver_closed:
+ if event.sender:
+ self.drained = event.sender.drained()
+ if self.drained == self.expected_drained:
+ self.error = None
+ self.sender_conn.close()
+ self.receiver_conn.close()
+
+ def run(self):
+ Container(self).run()
+
+
if __name__ == '__main__':
unittest.main(main_module())
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org