You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@qpid.apache.org by GitBox <gi...@apache.org> on 2021/07/16 19:38:45 UTC
[GitHub] [qpid-dispatch] ganeshmurthy opened a new pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
ganeshmurthy opened a new pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301
…control window updates
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] kgiusti commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
kgiusti commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r675802653
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -2496,10 +2657,21 @@ static void handle_connection_event(pn_event_t *e, qd_server_t *qd_server, void
}
case PN_RAW_CONNECTION_WAKE: {
qd_log(log, QD_LOG_TRACE, "[C%"PRIu64"] PN_RAW_CONNECTION_WAKE Wake-up", conn->conn_id);
+ if (IS_ATOMIC_FLAG_SET(&conn->q2_restart)) {
Review comment:
Precisely - SET and CLEAR operations must return the *previous* value, so if (CLEAR_ATOMIC_FLAG()) is true you know it went from True->False ... atomically.
We don't have a conditional test-and-set atomic operation.. yet.
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673609267
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -493,43 +559,32 @@ static int on_data_chunk_recv_callback(nghttp2_session *session,
// We want to be able to keep collecting the incoming DATA in the message object so we can ultimately route it when the credit does ultimately arrive.
//
if (stream_data->in_dlv || stream_data->header_and_props_composed) {
- if (!stream_data->stream_force_closed) {
- // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated that body
- // in the stream_data->body (in the else part). But before the rest of the HTTP2 data arrives, we got credit to send the delivery
- // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body and call qd_message_stream_data_append
- if (stream_data->body) {
- if (!stream_data->body_data_added) {
- qd_buffer_list_t existing_buffers;
- DEQ_INIT(existing_buffers);
- qd_compose_take_buffers(stream_data->body, &existing_buffers);
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &existing_buffers, 0);
- stream_data->body_data_added = true;
- }
- }
- else {
- // Add a dummy body so that other code that checks for the presense of stream_data->body will be satisfied.
- // This dummy body field will be be used and will not be sent.
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- stream_data->body_data_added = true;
- }
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &buffers, 0);
- qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] HTTP2 DATA on_data_chunk_recv_callback qd_compose_insert_binary_buffers into stream_data->message", conn->conn_id, stream_id);
- }
- else {
- qd_buffer_list_free_buffers(&buffers);
- }
+ qd_buffer_list_t buffers;
+ DEQ_INIT(buffers);
+ qd_buffer_list_append(&buffers, (uint8_t *)data, len);
+ // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated the body buffers
+ // in the stream_data->body_buffers. But before the rest of the HTTP2 data arrives, we got credit to send the delivery
+ // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body_buffers and call qd_message_stream_data_append
+
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
Review comment:
This is an issue with my editor which chuck pointed out this morning. 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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] kgiusti commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
kgiusti commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673417303
##########
File path: tests/system_tests_http2.py
##########
@@ -728,3 +741,126 @@ def test_goaway(self):
address = self.router_qdra.http_addresses[0] + "/goaway_test_1"
out = self.run_curl(address, args=["-i"])
self.assertIn("HTTP/2 503", out)
+
+
+class Http2Q2OneRouterTest(Http2TestBase):
+ @classmethod
+ def setUpClass(cls):
+ super(Http2Q2OneRouterTest, cls).setUpClass()
+ if skip_h2_test():
+ return
+ cls.http2_server_name = "http2_slow_q2_server"
+ os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port())
+ cls.http2_server = cls.tester.http2server(name=cls.http2_server_name,
+ listen_port=int(os.getenv('SERVER_LISTEN_PORT')),
+ py_string='python3',
+ server_file="http2_slow_q2_server.py")
+ name = "http2-test-router"
+ cls.connector_name = 'connectorToServer'
+ cls.connector_props = {
+ 'port': os.getenv('SERVER_LISTEN_PORT'),
+ 'address': 'examples',
+ 'host': '127.0.0.1',
+ 'protocolVersion': 'HTTP2',
+ 'name': cls.connector_name
+ }
+ config = Qdrouterd.Config([
+ ('router', {'mode': 'standalone', 'id': 'QDR'}),
+ ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}),
+
+ ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples',
+ 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}),
+ ('httpConnector', cls.connector_props)
+ ])
+ cls.router_qdra = cls.tester.qdrouterd(name, config, wait=True)
+
+ @unittest.skipIf(skip_h2_test(),
+ "Python 3.7 or greater, hyper-h2 and curl needed to run hyperhttp2 tests")
+ def test_q2_block_unblock(self):
+ # curl -X POST -H "Content-Type: multipart/form-data" -F "data=@/home/gmurthy/opensource/test.jpg"
+ # http://127.0.0.1:9000/upload --http2-prior-knowledge
+ address = self.router_qdra.http_addresses[0] + "/upload"
+ out = self.run_curl(address, args=['-X', 'POST', '-H', 'Content-Type: multipart/form-data',
+ '-F', 'data=@' + image_file('test.jpg')])
+ self.assertIn('Success', out)
+ num_blocked = 0
+ num_unblocked = 0
+ blocked = "blocked"
+ unblocked = "unblocked"
+ with open(self.router_qdra.logfile_path, 'r') as router_log:
+ log_lines = router_log.read().split("\n")
+ for log_line in log_lines:
+ if unblocked in log_line:
+ num_unblocked += 1
+ elif blocked in log_line:
+ num_blocked += 1
+
+ self.assertGreater(num_blocked, 0)
+ self.assertGreater(num_unblocked, 0)
+
+
+class Http2Q2TwoRouterTest(Http2TestBase):
+ @classmethod
+ def setUpClass(cls):
+ super(Http2Q2TwoRouterTest, cls).setUpClass()
+ if skip_h2_test():
+ return
+ cls.http2_server_name = "http2_server"
+ os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port())
+ cls.http2_server = cls.tester.http2server(name=cls.http2_server_name,
+ listen_port=int(os.getenv('SERVER_LISTEN_PORT')),
+ py_string='python3',
+ server_file="http2_slow_q2_server.py")
+ qdr_a = "QDR.A"
+ inter_router_port = cls.tester.get_port()
+ config_qdra = Qdrouterd.Config([
+ ('router', {'mode': 'interior', 'id': 'QDR.A'}),
+ ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}),
+ ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples',
+ 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}),
+ ('connector', {'name': 'connectorToB', 'role': 'inter-router',
+ 'port': inter_router_port,
+ 'verifyHostname': 'no'})
+ ])
+
+ qdr_b = "QDR.B"
+ cls.connector_name = 'serverConnector'
+ cls.http_connector_props = {
+ 'port': os.getenv('SERVER_LISTEN_PORT'),
+ 'address': 'examples',
+ 'host': '127.0.0.1',
+ 'protocolVersion': 'HTTP2',
+ 'name': cls.connector_name
+ }
+ config_qdrb = Qdrouterd.Config([
+ ('router', {'mode': 'interior', 'id': 'QDR.B'}),
+ ('httpConnector', cls.http_connector_props),
+ ('listener', {'role': 'inter-router', 'maxSessionFrames': '10', 'port': inter_router_port})
+ ])
+ cls.router_qdrb = cls.tester.qdrouterd(qdr_b, config_qdrb, wait=True)
+ cls.router_qdra = cls.tester.qdrouterd(qdr_a, config_qdra, wait=True)
+ cls.router_qdra.wait_router_connected('QDR.B')
+
+ @unittest.skipIf(skip_h2_test(),
+ "Python 3.7 or greater, hyper-h2 and curl needed to run hyperhttp2 tests")
+ def test_q2_block_unblock(self):
+ # curl -X POST -H "Content-Type: multipart/form-data" -F "data=@/home/gmurthy/opensource/test.jpg"
+ # http://127.0.0.1:9000/upload --http2-prior-knowledge
+ address = self.router_qdra.http_addresses[0] + "/upload"
+ out = self.run_curl(address, args=['-X', 'POST', '-H', 'Content-Type: multipart/form-data',
+ '-F', 'data=@' + image_file('test.jpg')])
+ self.assertIn('Success', out)
+ num_blocked = 0
+ num_unblocked = 0
+ blocked = "blocked"
+ unblocked = "unblocked"
+ with open(self.router_qdra.logfile_path, 'r') as router_log:
+ log_lines = router_log.read().split("\n")
+ for log_line in log_lines:
+ if unblocked in log_line:
+ num_unblocked += 1
+ elif blocked in log_line:
+ num_blocked += 1
+
+ self.assertGreater(num_blocked, 0)
Review comment:
same here num_blocked == num_unblocked
##########
File path: src/message.c
##########
@@ -2973,18 +2981,36 @@ bool qd_message_oversize(const qd_message_t *msg)
}
+int qd_message_stream_data_footer_append(qd_message_t *message, qd_buffer_list_t *footer_props)
+{
+ qd_composed_field_t *field = 0;
+ int rc = 0;
+
+ field = qd_compose(QD_PERFORMATIVE_FOOTER, field);
+
+ // Stick the buffers into the footer compose field.
+ qd_compose_insert_binary_buffers(field, footer_props);
+
+ // This is the footer, we don't care about q2 and will disable it.
+ qd_message_Q2_holdoff_disable(message);
+
+ rc = qd_message_extend(message, field, 0);
+
+ qd_compose_free(field);
+ return rc;
+
+}
+
int qd_message_stream_data_append(qd_message_t *message, qd_buffer_list_t *data, bool *q2_blocked)
{
unsigned int length = DEQ_SIZE(*data);
+
qd_composed_field_t *field = 0;
int rc = 0;
if (q2_blocked)
*q2_blocked = false;
- if (length == 0)
Review comment:
Why is inserting an empty body buffer necessary?
##########
File path: src/message.c
##########
@@ -2973,18 +2981,36 @@ bool qd_message_oversize(const qd_message_t *msg)
}
+int qd_message_stream_data_footer_append(qd_message_t *message, qd_buffer_list_t *footer_props)
+{
+ qd_composed_field_t *field = 0;
+ int rc = 0;
+
+ field = qd_compose(QD_PERFORMATIVE_FOOTER, field);
+
+ // Stick the buffers into the footer compose field.
+ qd_compose_insert_binary_buffers(field, footer_props);
+
+ // This is the footer, we don't care about q2 and will disable it.
+ qd_message_Q2_holdoff_disable(message);
Review comment:
Wait, you're building a amqp message, so I'm assuming this message will be sent - why would a _sender_ disable Q2 on an outgoing message?
I can see a _receiver_ (amqp->http2_adaptor->endpoint) disable Q2 if it expected, I dunno, a footer section to be bigger than Q2... but a sender?
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -342,14 +369,16 @@ static void free_http2_stream_data(qdr_http2_stream_data_t *stream_data, bool on
}
free(stream_data->reply_to);
qd_compose_free(stream_data->app_properties);
- qd_compose_free(stream_data->body);
+ qd_buffer_list_free_buffers(&stream_data->body_buffers);
qd_compose_free(stream_data->footer_properties);
if (DEQ_SIZE(session_data->streams) > 0) {
DEQ_REMOVE(session_data->streams, stream_data);
nghttp2_session_set_stream_user_data(session_data->session, stream_data->stream_id, NULL);
}
- if (stream_data->method) free(stream_data->method);
- if (stream_data->remote_site) free(stream_data->remote_site);
+ if (stream_data->method)
+ free(stream_data->method);
Review comment:
FYI: The free() function can take a null pointer - it will check it for you, so you don't have to do if(x) free(x), you can just free(x)
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -493,43 +559,32 @@ static int on_data_chunk_recv_callback(nghttp2_session *session,
// We want to be able to keep collecting the incoming DATA in the message object so we can ultimately route it when the credit does ultimately arrive.
//
if (stream_data->in_dlv || stream_data->header_and_props_composed) {
- if (!stream_data->stream_force_closed) {
- // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated that body
- // in the stream_data->body (in the else part). But before the rest of the HTTP2 data arrives, we got credit to send the delivery
- // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body and call qd_message_stream_data_append
- if (stream_data->body) {
- if (!stream_data->body_data_added) {
- qd_buffer_list_t existing_buffers;
- DEQ_INIT(existing_buffers);
- qd_compose_take_buffers(stream_data->body, &existing_buffers);
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &existing_buffers, 0);
- stream_data->body_data_added = true;
- }
- }
- else {
- // Add a dummy body so that other code that checks for the presense of stream_data->body will be satisfied.
- // This dummy body field will be be used and will not be sent.
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- stream_data->body_data_added = true;
- }
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &buffers, 0);
- qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] HTTP2 DATA on_data_chunk_recv_callback qd_compose_insert_binary_buffers into stream_data->message", conn->conn_id, stream_id);
- }
- else {
- qd_buffer_list_free_buffers(&buffers);
- }
+ qd_buffer_list_t buffers;
+ DEQ_INIT(buffers);
+ qd_buffer_list_append(&buffers, (uint8_t *)data, len);
+ // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated the body buffers
+ // in the stream_data->body_buffers. But before the rest of the HTTP2 data arrives, we got credit to send the delivery
+ // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body_buffers and call qd_message_stream_data_append
+
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
+ if (!stream_data->body_data_added_to_msg) {
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, 0);
Review comment:
Why is Q2 blocking not checked for here? If it is expected to never trigger Q2 I would recommend at least passing a temp bool to it and doing an assert(temp == false) to ensure no future maintainers break that.
##########
File path: tests/system_tests_http2.py
##########
@@ -728,3 +741,126 @@ def test_goaway(self):
address = self.router_qdra.http_addresses[0] + "/goaway_test_1"
out = self.run_curl(address, args=["-i"])
self.assertIn("HTTP/2 503", out)
+
+
+class Http2Q2OneRouterTest(Http2TestBase):
+ @classmethod
+ def setUpClass(cls):
+ super(Http2Q2OneRouterTest, cls).setUpClass()
+ if skip_h2_test():
+ return
+ cls.http2_server_name = "http2_slow_q2_server"
+ os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port())
+ cls.http2_server = cls.tester.http2server(name=cls.http2_server_name,
+ listen_port=int(os.getenv('SERVER_LISTEN_PORT')),
+ py_string='python3',
+ server_file="http2_slow_q2_server.py")
+ name = "http2-test-router"
+ cls.connector_name = 'connectorToServer'
+ cls.connector_props = {
+ 'port': os.getenv('SERVER_LISTEN_PORT'),
+ 'address': 'examples',
+ 'host': '127.0.0.1',
+ 'protocolVersion': 'HTTP2',
+ 'name': cls.connector_name
+ }
+ config = Qdrouterd.Config([
+ ('router', {'mode': 'standalone', 'id': 'QDR'}),
+ ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}),
+
+ ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples',
+ 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}),
+ ('httpConnector', cls.connector_props)
+ ])
+ cls.router_qdra = cls.tester.qdrouterd(name, config, wait=True)
+
+ @unittest.skipIf(skip_h2_test(),
+ "Python 3.7 or greater, hyper-h2 and curl needed to run hyperhttp2 tests")
+ def test_q2_block_unblock(self):
+ # curl -X POST -H "Content-Type: multipart/form-data" -F "data=@/home/gmurthy/opensource/test.jpg"
+ # http://127.0.0.1:9000/upload --http2-prior-knowledge
+ address = self.router_qdra.http_addresses[0] + "/upload"
+ out = self.run_curl(address, args=['-X', 'POST', '-H', 'Content-Type: multipart/form-data',
+ '-F', 'data=@' + image_file('test.jpg')])
+ self.assertIn('Success', out)
+ num_blocked = 0
+ num_unblocked = 0
+ blocked = "blocked"
+ unblocked = "unblocked"
+ with open(self.router_qdra.logfile_path, 'r') as router_log:
+ log_lines = router_log.read().split("\n")
+ for log_line in log_lines:
+ if unblocked in log_line:
+ num_unblocked += 1
+ elif blocked in log_line:
+ num_blocked += 1
+
+ self.assertGreater(num_blocked, 0)
+ self.assertGreater(num_unblocked, 0)
Review comment:
shouldn't num_blocked == num_unblocked as well? HTTP/1 tests for that and it found a bug where unblock never happened
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -770,63 +856,102 @@ static bool compose_and_deliver(qdr_http2_connection_t *conn, qdr_http2_stream_d
if (!stream_data->header_and_props_composed) {
qd_composed_field_t *header_and_props = 0;
if (conn->ingress) {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- conn->config->address, // const char *to
- 0, // const char *subject
- stream_data->reply_to, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ conn->config->address, // const char *to
+ 0, // const char *subject
+ stream_data->reply_to, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
else {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- stream_data->reply_to, // const char *to
- 0, // const char *subject
- 0, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ stream_data->reply_to, // const char *to
+ 0, // const char *subject
+ 0, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
if (receive_complete) {
qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = true in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
- if (!stream_data->body) {
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_compose_insert_binary(stream_data->body, 0, 0);
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] Inserting empty body data in compose_and_deliver", conn->conn_id, stream_data->stream_id);
- }
-
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
+
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
}
+
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
- if (stream_data->body) {
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
+ qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body_buffers in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
}
- stream_data->body_data_added = true;
+ stream_data->body_data_added_to_msg = true;
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
-
if (stream_data->footer_properties) {
+
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
//
// The footer has already arrived but there was no body. Insert an empty body
//
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, 0);
Review comment:
ditto - no check for q2 activation
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -359,12 +388,34 @@ static void free_http2_stream_data(qdr_http2_stream_data_t *stream_data, bool on
qd_message_free(stream_data->message);
}
+ //
+ // If the client/server closed the connection abruptly, we need to release the stream_data->curr_stream_data and
+ // stream_data->next_stream_data.
+ // This final decref of the delivery is going to free the associated message but before this message can be freed
+ // all stream data (body data) objects need to be freed. We do this here.
+ //
if (stream_data->in_dlv && !stream_data->in_dlv_decrefed) {
+ if (stream_data->curr_stream_data) {
Review comment:
FYI: Whenever possible the dispatch code base adopts the same behavior as "free()" when it comes to releasing pointers to resources: qd_message_stream_data_release(0) is allowed.
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -2496,10 +2657,21 @@ static void handle_connection_event(pn_event_t *e, qd_server_t *qd_server, void
}
case PN_RAW_CONNECTION_WAKE: {
qd_log(log, QD_LOG_TRACE, "[C%"PRIu64"] PN_RAW_CONNECTION_WAKE Wake-up", conn->conn_id);
+ if (IS_ATOMIC_FLAG_SET(&conn->q2_restart)) {
Review comment:
This needs to be an atomic set and clear, otherwise you'll miss transitions.
Use CLEAR_ATOMIC_FLAG and check the return value for true.
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -493,43 +559,32 @@ static int on_data_chunk_recv_callback(nghttp2_session *session,
// We want to be able to keep collecting the incoming DATA in the message object so we can ultimately route it when the credit does ultimately arrive.
//
if (stream_data->in_dlv || stream_data->header_and_props_composed) {
- if (!stream_data->stream_force_closed) {
- // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated that body
- // in the stream_data->body (in the else part). But before the rest of the HTTP2 data arrives, we got credit to send the delivery
- // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body and call qd_message_stream_data_append
- if (stream_data->body) {
- if (!stream_data->body_data_added) {
- qd_buffer_list_t existing_buffers;
- DEQ_INIT(existing_buffers);
- qd_compose_take_buffers(stream_data->body, &existing_buffers);
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &existing_buffers, 0);
- stream_data->body_data_added = true;
- }
- }
- else {
- // Add a dummy body so that other code that checks for the presense of stream_data->body will be satisfied.
- // This dummy body field will be be used and will not be sent.
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- stream_data->body_data_added = true;
- }
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &buffers, 0);
- qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] HTTP2 DATA on_data_chunk_recv_callback qd_compose_insert_binary_buffers into stream_data->message", conn->conn_id, stream_id);
- }
- else {
- qd_buffer_list_free_buffers(&buffers);
- }
+ qd_buffer_list_t buffers;
+ DEQ_INIT(buffers);
+ qd_buffer_list_append(&buffers, (uint8_t *)data, len);
+ // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated the body buffers
+ // in the stream_data->body_buffers. But before the rest of the HTTP2 data arrives, we got credit to send the delivery
+ // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body_buffers and call qd_message_stream_data_append
+
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
Review comment:
The indentation is not showing up correctly. Is that github, or are you mistakenly using tabs instead of 4 spaces?
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] codecov-commenter commented on pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#issuecomment-882163509
# [Codecov](https://codecov.io/gh/apache/qpid-dispatch/pull/1301?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#1301](https://codecov.io/gh/apache/qpid-dispatch/pull/1301?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (22b9b36) into [main](https://codecov.io/gh/apache/qpid-dispatch/commit/b1c25e94c76926c35ee48ced4053abfc23fdf347?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b1c25e9) will **increase** coverage by `0.00%`.
> The diff coverage is `18.75%`.
[](https://codecov.io/gh/apache/qpid-dispatch/pull/1301?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## main #1301 +/- ##
=======================================
Coverage 84.60% 84.61%
=======================================
Files 113 113
Lines 28326 28337 +11
=======================================
+ Hits 23965 23977 +12
+ Misses 4361 4360 -1
```
| [Impacted Files](https://codecov.io/gh/apache/qpid-dispatch/pull/1301?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [src/message.c](https://codecov.io/gh/apache/qpid-dispatch/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3JjL21lc3NhZ2UuYw==) | `87.23% <18.75%> (-0.72%)` | :arrow_down: |
| [src/container.c](https://codecov.io/gh/apache/qpid-dispatch/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3JjL2NvbnRhaW5lci5j) | `84.41% <0.00%> (-0.03%)` | :arrow_down: |
| [src/router\_node.c](https://codecov.io/gh/apache/qpid-dispatch/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3JjL3JvdXRlcl9ub2RlLmM=) | `93.43% <0.00%> (-0.02%)` | :arrow_down: |
| [src/parse.c](https://codecov.io/gh/apache/qpid-dispatch/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3JjL3BhcnNlLmM=) | `88.18% <0.00%> (+0.21%)` | :arrow_up: |
| [src/router\_core/connections.c](https://codecov.io/gh/apache/qpid-dispatch/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3JjL3JvdXRlcl9jb3JlL2Nvbm5lY3Rpb25zLmM=) | `89.75% <0.00%> (+0.29%)` | :arrow_up: |
| [src/router\_core/transfer.c](https://codecov.io/gh/apache/qpid-dispatch/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3JjL3JvdXRlcl9jb3JlL3RyYW5zZmVyLmM=) | `94.18% <0.00%> (+0.64%)` | :arrow_up: |
| [src/router\_core/router\_core.c](https://codecov.io/gh/apache/qpid-dispatch/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3JjL3JvdXRlcl9jb3JlL3JvdXRlcl9jb3JlLmM=) | `87.04% <0.00%> (+0.96%)` | :arrow_up: |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/qpid-dispatch/pull/1301?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/qpid-dispatch/pull/1301?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [b1c25e9...22b9b36](https://codecov.io/gh/apache/qpid-dispatch/pull/1301?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] kgiusti commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
kgiusti commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r675802653
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -2496,10 +2657,21 @@ static void handle_connection_event(pn_event_t *e, qd_server_t *qd_server, void
}
case PN_RAW_CONNECTION_WAKE: {
qd_log(log, QD_LOG_TRACE, "[C%"PRIu64"] PN_RAW_CONNECTION_WAKE Wake-up", conn->conn_id);
+ if (IS_ATOMIC_FLAG_SET(&conn->q2_restart)) {
Review comment:
Precisely - SET and CLEAR operations must return the *previous* value, so if (CLEAR_ATOMIC_FLAG()) is true you know it went from True->False ... atomically.
Imagine two threads running this code concurrently, one sees IS_ATOMIC_FLAG_SET() as true, enters the IF statement, then immediately swaps out. The second thread also sees IS_ATOMIC_FLAG_SET() as true and also enters the IF block. With an atomic CLEAR->return old value it is guaranteed that only one thread will ever see the return value of True and enter the IF.
That said, it won't make a difference since only one thread is running the event handler so you're pretty much guaranteed that no other thread will be making the same IS_XXX CLEAR__XX calls concurrently.
But why do two atomic operations (both of which cause a cache flush) when one call does both?
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] asfgit closed pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673609341
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -493,43 +559,32 @@ static int on_data_chunk_recv_callback(nghttp2_session *session,
// We want to be able to keep collecting the incoming DATA in the message object so we can ultimately route it when the credit does ultimately arrive.
//
if (stream_data->in_dlv || stream_data->header_and_props_composed) {
- if (!stream_data->stream_force_closed) {
- // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated that body
- // in the stream_data->body (in the else part). But before the rest of the HTTP2 data arrives, we got credit to send the delivery
- // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body and call qd_message_stream_data_append
- if (stream_data->body) {
- if (!stream_data->body_data_added) {
- qd_buffer_list_t existing_buffers;
- DEQ_INIT(existing_buffers);
- qd_compose_take_buffers(stream_data->body, &existing_buffers);
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &existing_buffers, 0);
- stream_data->body_data_added = true;
- }
- }
- else {
- // Add a dummy body so that other code that checks for the presense of stream_data->body will be satisfied.
- // This dummy body field will be be used and will not be sent.
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- stream_data->body_data_added = true;
- }
- // @TODO(kgiusti): handle Q2 block event:
- qd_message_stream_data_append(stream_data->message, &buffers, 0);
- qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] HTTP2 DATA on_data_chunk_recv_callback qd_compose_insert_binary_buffers into stream_data->message", conn->conn_id, stream_id);
- }
- else {
- qd_buffer_list_free_buffers(&buffers);
- }
+ qd_buffer_list_t buffers;
+ DEQ_INIT(buffers);
+ qd_buffer_list_append(&buffers, (uint8_t *)data, len);
+ // DISPATCH-1868: Part of the HTTP2 message body arrives *before* we can route the delivery. So we accumulated the body buffers
+ // in the stream_data->body_buffers. But before the rest of the HTTP2 data arrives, we got credit to send the delivery
+ // and we have an in_dlv object now. Now, we take the buffers that were added previously to stream_data->body_buffers and call qd_message_stream_data_append
+
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
+ if (!stream_data->body_data_added_to_msg) {
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, 0);
Review comment:
Added q2 checking 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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] kgiusti commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
kgiusti commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r675789529
##########
File path: src/message.c
##########
@@ -2973,18 +2981,36 @@ bool qd_message_oversize(const qd_message_t *msg)
}
+int qd_message_stream_data_footer_append(qd_message_t *message, qd_buffer_list_t *footer_props)
+{
+ qd_composed_field_t *field = 0;
+ int rc = 0;
+
+ field = qd_compose(QD_PERFORMATIVE_FOOTER, field);
+
+ // Stick the buffers into the footer compose field.
+ qd_compose_insert_binary_buffers(field, footer_props);
+
+ // This is the footer, we don't care about q2 and will disable it.
+ qd_message_Q2_holdoff_disable(message);
+
+ rc = qd_message_extend(message, field, 0);
+
+ qd_compose_free(field);
+ return rc;
+
+}
+
int qd_message_stream_data_append(qd_message_t *message, qd_buffer_list_t *data, bool *q2_blocked)
{
unsigned int length = DEQ_SIZE(*data);
+
qd_composed_field_t *field = 0;
int rc = 0;
if (q2_blocked)
*q2_blocked = false;
- if (length == 0)
Review comment:
You're correct - the message has to have *something* in the body, so yeah if we're trying to send bodyless meta-data this looks good.
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673606660
##########
File path: tests/system_tests_http2.py
##########
@@ -728,3 +741,126 @@ def test_goaway(self):
address = self.router_qdra.http_addresses[0] + "/goaway_test_1"
out = self.run_curl(address, args=["-i"])
self.assertIn("HTTP/2 503", out)
+
+
+class Http2Q2OneRouterTest(Http2TestBase):
+ @classmethod
+ def setUpClass(cls):
+ super(Http2Q2OneRouterTest, cls).setUpClass()
+ if skip_h2_test():
+ return
+ cls.http2_server_name = "http2_slow_q2_server"
+ os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port())
+ cls.http2_server = cls.tester.http2server(name=cls.http2_server_name,
+ listen_port=int(os.getenv('SERVER_LISTEN_PORT')),
+ py_string='python3',
+ server_file="http2_slow_q2_server.py")
+ name = "http2-test-router"
+ cls.connector_name = 'connectorToServer'
+ cls.connector_props = {
+ 'port': os.getenv('SERVER_LISTEN_PORT'),
+ 'address': 'examples',
+ 'host': '127.0.0.1',
+ 'protocolVersion': 'HTTP2',
+ 'name': cls.connector_name
+ }
+ config = Qdrouterd.Config([
+ ('router', {'mode': 'standalone', 'id': 'QDR'}),
+ ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}),
+
+ ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples',
+ 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}),
+ ('httpConnector', cls.connector_props)
+ ])
+ cls.router_qdra = cls.tester.qdrouterd(name, config, wait=True)
+
+ @unittest.skipIf(skip_h2_test(),
+ "Python 3.7 or greater, hyper-h2 and curl needed to run hyperhttp2 tests")
+ def test_q2_block_unblock(self):
+ # curl -X POST -H "Content-Type: multipart/form-data" -F "data=@/home/gmurthy/opensource/test.jpg"
+ # http://127.0.0.1:9000/upload --http2-prior-knowledge
+ address = self.router_qdra.http_addresses[0] + "/upload"
+ out = self.run_curl(address, args=['-X', 'POST', '-H', 'Content-Type: multipart/form-data',
+ '-F', 'data=@' + image_file('test.jpg')])
+ self.assertIn('Success', out)
+ num_blocked = 0
+ num_unblocked = 0
+ blocked = "blocked"
+ unblocked = "unblocked"
+ with open(self.router_qdra.logfile_path, 'r') as router_log:
+ log_lines = router_log.read().split("\n")
+ for log_line in log_lines:
+ if unblocked in log_line:
+ num_unblocked += 1
+ elif blocked in log_line:
+ num_blocked += 1
+
+ self.assertGreater(num_blocked, 0)
+ self.assertGreater(num_unblocked, 0)
Review comment:
In http2 the size of the buffers given to proton is 16k each and we give 4 of those to proton. Sometimes we end up giving as much as 128k to proton before the first 64 bytes can be decoded by nghttp2 and it starts calling on_data_chunk_recv_callback(). So there are several calls made to on_data_chunk_recv_callback() where the q2 limit log statement is written. So cannot really match num_blocked and num_unblocked
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673610009
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -2496,10 +2657,21 @@ static void handle_connection_event(pn_event_t *e, qd_server_t *qd_server, void
}
case PN_RAW_CONNECTION_WAKE: {
qd_log(log, QD_LOG_TRACE, "[C%"PRIu64"] PN_RAW_CONNECTION_WAKE Wake-up", conn->conn_id);
+ if (IS_ATOMIC_FLAG_SET(&conn->q2_restart)) {
Review comment:
Added CLEAR_ATOMIC_FLAG. Check the return value of CLEAR_ATOMIC_FLAG ?
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673609421
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -342,14 +369,16 @@ static void free_http2_stream_data(qdr_http2_stream_data_t *stream_data, bool on
}
free(stream_data->reply_to);
qd_compose_free(stream_data->app_properties);
- qd_compose_free(stream_data->body);
+ qd_buffer_list_free_buffers(&stream_data->body_buffers);
qd_compose_free(stream_data->footer_properties);
if (DEQ_SIZE(session_data->streams) > 0) {
DEQ_REMOVE(session_data->streams, stream_data);
nghttp2_session_set_stream_user_data(session_data->session, stream_data->stream_id, NULL);
}
- if (stream_data->method) free(stream_data->method);
- if (stream_data->remote_site) free(stream_data->remote_site);
+ if (stream_data->method)
+ free(stream_data->method);
Review comment:
Done
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -359,12 +388,34 @@ static void free_http2_stream_data(qdr_http2_stream_data_t *stream_data, bool on
qd_message_free(stream_data->message);
}
+ //
+ // If the client/server closed the connection abruptly, we need to release the stream_data->curr_stream_data and
+ // stream_data->next_stream_data.
+ // This final decref of the delivery is going to free the associated message but before this message can be freed
+ // all stream data (body data) objects need to be freed. We do this here.
+ //
if (stream_data->in_dlv && !stream_data->in_dlv_decrefed) {
+ if (stream_data->curr_stream_data) {
Review comment:
Done
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673971868
##########
File path: src/message.c
##########
@@ -2973,18 +2981,36 @@ bool qd_message_oversize(const qd_message_t *msg)
}
+int qd_message_stream_data_footer_append(qd_message_t *message, qd_buffer_list_t *footer_props)
+{
+ qd_composed_field_t *field = 0;
+ int rc = 0;
+
+ field = qd_compose(QD_PERFORMATIVE_FOOTER, field);
+
+ // Stick the buffers into the footer compose field.
+ qd_compose_insert_binary_buffers(field, footer_props);
+
+ // This is the footer, we don't care about q2 and will disable it.
+ qd_message_Q2_holdoff_disable(message);
+
+ rc = qd_message_extend(message, field, 0);
+
+ qd_compose_free(field);
+ return rc;
+
+}
+
int qd_message_stream_data_append(qd_message_t *message, qd_buffer_list_t *data, bool *q2_blocked)
{
unsigned int length = DEQ_SIZE(*data);
+
qd_composed_field_t *field = 0;
int rc = 0;
if (q2_blocked)
*q2_blocked = false;
- if (length == 0)
Review comment:
It is my understanding that an AMQP message needs to at least have an empty QD_PERFORMATIVE_BODY_DATA ?
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673610363
##########
File path: tests/system_tests_http2.py
##########
@@ -728,3 +741,126 @@ def test_goaway(self):
address = self.router_qdra.http_addresses[0] + "/goaway_test_1"
out = self.run_curl(address, args=["-i"])
self.assertIn("HTTP/2 503", out)
+
+
+class Http2Q2OneRouterTest(Http2TestBase):
+ @classmethod
+ def setUpClass(cls):
+ super(Http2Q2OneRouterTest, cls).setUpClass()
+ if skip_h2_test():
+ return
+ cls.http2_server_name = "http2_slow_q2_server"
+ os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port())
+ cls.http2_server = cls.tester.http2server(name=cls.http2_server_name,
+ listen_port=int(os.getenv('SERVER_LISTEN_PORT')),
+ py_string='python3',
+ server_file="http2_slow_q2_server.py")
+ name = "http2-test-router"
+ cls.connector_name = 'connectorToServer'
+ cls.connector_props = {
+ 'port': os.getenv('SERVER_LISTEN_PORT'),
+ 'address': 'examples',
+ 'host': '127.0.0.1',
+ 'protocolVersion': 'HTTP2',
+ 'name': cls.connector_name
+ }
+ config = Qdrouterd.Config([
+ ('router', {'mode': 'standalone', 'id': 'QDR'}),
+ ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}),
+
+ ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples',
+ 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}),
+ ('httpConnector', cls.connector_props)
+ ])
+ cls.router_qdra = cls.tester.qdrouterd(name, config, wait=True)
+
+ @unittest.skipIf(skip_h2_test(),
+ "Python 3.7 or greater, hyper-h2 and curl needed to run hyperhttp2 tests")
+ def test_q2_block_unblock(self):
+ # curl -X POST -H "Content-Type: multipart/form-data" -F "data=@/home/gmurthy/opensource/test.jpg"
+ # http://127.0.0.1:9000/upload --http2-prior-knowledge
+ address = self.router_qdra.http_addresses[0] + "/upload"
+ out = self.run_curl(address, args=['-X', 'POST', '-H', 'Content-Type: multipart/form-data',
+ '-F', 'data=@' + image_file('test.jpg')])
+ self.assertIn('Success', out)
+ num_blocked = 0
+ num_unblocked = 0
+ blocked = "blocked"
+ unblocked = "unblocked"
+ with open(self.router_qdra.logfile_path, 'r') as router_log:
+ log_lines = router_log.read().split("\n")
+ for log_line in log_lines:
+ if unblocked in log_line:
+ num_unblocked += 1
+ elif blocked in log_line:
+ num_blocked += 1
+
+ self.assertGreater(num_blocked, 0)
+ self.assertGreater(num_unblocked, 0)
+
+
+class Http2Q2TwoRouterTest(Http2TestBase):
+ @classmethod
+ def setUpClass(cls):
+ super(Http2Q2TwoRouterTest, cls).setUpClass()
+ if skip_h2_test():
+ return
+ cls.http2_server_name = "http2_server"
+ os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port())
+ cls.http2_server = cls.tester.http2server(name=cls.http2_server_name,
+ listen_port=int(os.getenv('SERVER_LISTEN_PORT')),
+ py_string='python3',
+ server_file="http2_slow_q2_server.py")
+ qdr_a = "QDR.A"
+ inter_router_port = cls.tester.get_port()
+ config_qdra = Qdrouterd.Config([
+ ('router', {'mode': 'interior', 'id': 'QDR.A'}),
+ ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}),
+ ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples',
+ 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}),
+ ('connector', {'name': 'connectorToB', 'role': 'inter-router',
+ 'port': inter_router_port,
+ 'verifyHostname': 'no'})
+ ])
+
+ qdr_b = "QDR.B"
+ cls.connector_name = 'serverConnector'
+ cls.http_connector_props = {
+ 'port': os.getenv('SERVER_LISTEN_PORT'),
+ 'address': 'examples',
+ 'host': '127.0.0.1',
+ 'protocolVersion': 'HTTP2',
+ 'name': cls.connector_name
+ }
+ config_qdrb = Qdrouterd.Config([
+ ('router', {'mode': 'interior', 'id': 'QDR.B'}),
+ ('httpConnector', cls.http_connector_props),
+ ('listener', {'role': 'inter-router', 'maxSessionFrames': '10', 'port': inter_router_port})
+ ])
+ cls.router_qdrb = cls.tester.qdrouterd(qdr_b, config_qdrb, wait=True)
+ cls.router_qdra = cls.tester.qdrouterd(qdr_a, config_qdra, wait=True)
+ cls.router_qdra.wait_router_connected('QDR.B')
+
+ @unittest.skipIf(skip_h2_test(),
+ "Python 3.7 or greater, hyper-h2 and curl needed to run hyperhttp2 tests")
+ def test_q2_block_unblock(self):
+ # curl -X POST -H "Content-Type: multipart/form-data" -F "data=@/home/gmurthy/opensource/test.jpg"
+ # http://127.0.0.1:9000/upload --http2-prior-knowledge
+ address = self.router_qdra.http_addresses[0] + "/upload"
+ out = self.run_curl(address, args=['-X', 'POST', '-H', 'Content-Type: multipart/form-data',
+ '-F', 'data=@' + image_file('test.jpg')])
+ self.assertIn('Success', out)
+ num_blocked = 0
+ num_unblocked = 0
+ blocked = "blocked"
+ unblocked = "unblocked"
+ with open(self.router_qdra.logfile_path, 'r') as router_log:
+ log_lines = router_log.read().split("\n")
+ for log_line in log_lines:
+ if unblocked in log_line:
+ num_unblocked += 1
+ elif blocked in log_line:
+ num_blocked += 1
+
+ self.assertGreater(num_blocked, 0)
Review comment:
Same response as above. The num_blocked and num_unblocked cannot match
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673484087
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -770,63 +856,102 @@ static bool compose_and_deliver(qdr_http2_connection_t *conn, qdr_http2_stream_d
if (!stream_data->header_and_props_composed) {
qd_composed_field_t *header_and_props = 0;
if (conn->ingress) {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- conn->config->address, // const char *to
- 0, // const char *subject
- stream_data->reply_to, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ conn->config->address, // const char *to
+ 0, // const char *subject
+ stream_data->reply_to, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
else {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- stream_data->reply_to, // const char *to
- 0, // const char *subject
- 0, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ stream_data->reply_to, // const char *to
+ 0, // const char *subject
+ 0, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
if (receive_complete) {
qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = true in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
- if (!stream_data->body) {
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_compose_insert_binary(stream_data->body, 0, 0);
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] Inserting empty body data in compose_and_deliver", conn->conn_id, stream_data->stream_id);
- }
-
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
+
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
}
+
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
- if (stream_data->body) {
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
+ qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body_buffers in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
}
- stream_data->body_data_added = true;
+ stream_data->body_data_added_to_msg = true;
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
-
if (stream_data->footer_properties) {
+
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
//
// The footer has already arrived but there was no body. Insert an empty body
//
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, 0);
Review comment:
In this specific case the entire footer has arrived but there is no body (DEQ_SIZE(stream_data->body_buffers) == 0)
We are simply calling the qd_message_stream_data_append() to insert an empty body. This will not trigger q2. Then we call qd_message_stream_data_footer_append() to append the footer.
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r675799980
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -770,63 +856,102 @@ static bool compose_and_deliver(qdr_http2_connection_t *conn, qdr_http2_stream_d
if (!stream_data->header_and_props_composed) {
qd_composed_field_t *header_and_props = 0;
if (conn->ingress) {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- conn->config->address, // const char *to
- 0, // const char *subject
- stream_data->reply_to, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ conn->config->address, // const char *to
+ 0, // const char *subject
+ stream_data->reply_to, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
else {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- stream_data->reply_to, // const char *to
- 0, // const char *subject
- 0, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ stream_data->reply_to, // const char *to
+ 0, // const char *subject
+ 0, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
if (receive_complete) {
qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = true in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
- if (!stream_data->body) {
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_compose_insert_binary(stream_data->body, 0, 0);
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] Inserting empty body data in compose_and_deliver", conn->conn_id, stream_data->stream_id);
- }
-
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
+
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
}
+
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
- if (stream_data->body) {
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
+ qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body_buffers in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
}
- stream_data->body_data_added = true;
+ stream_data->body_data_added_to_msg = true;
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
-
if (stream_data->footer_properties) {
+
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
//
// The footer has already arrived but there was no body. Insert an empty body
//
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, 0);
Review comment:
I will add the assert, thanks
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] ganeshmurthy commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
ganeshmurthy commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r673610121
##########
File path: src/message.c
##########
@@ -2973,18 +2981,36 @@ bool qd_message_oversize(const qd_message_t *msg)
}
+int qd_message_stream_data_footer_append(qd_message_t *message, qd_buffer_list_t *footer_props)
+{
+ qd_composed_field_t *field = 0;
+ int rc = 0;
+
+ field = qd_compose(QD_PERFORMATIVE_FOOTER, field);
+
+ // Stick the buffers into the footer compose field.
+ qd_compose_insert_binary_buffers(field, footer_props);
+
+ // This is the footer, we don't care about q2 and will disable it.
+ qd_message_Q2_holdoff_disable(message);
Review comment:
Agreed! Removed q2 disable call
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] kgiusti commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
kgiusti commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r675798329
##########
File path: src/adaptors/http2/http2_adaptor.c
##########
@@ -770,63 +856,102 @@ static bool compose_and_deliver(qdr_http2_connection_t *conn, qdr_http2_stream_d
if (!stream_data->header_and_props_composed) {
qd_composed_field_t *header_and_props = 0;
if (conn->ingress) {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- conn->config->address, // const char *to
- 0, // const char *subject
- stream_data->reply_to, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ conn->config->address, // const char *to
+ 0, // const char *subject
+ stream_data->reply_to, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
else {
- header_and_props = qd_message_compose_amqp(stream_data->message,
- stream_data->reply_to, // const char *to
- 0, // const char *subject
- 0, // const char *reply_to
- 0, // const char *content_type
- 0, // const char *content_encoding
- 0, // int32_t correlation_id
- conn->config->site);
+ header_and_props = qd_message_compose_amqp(conn,
+ stream_data->message,
+ stream_data->reply_to, // const char *to
+ 0, // const char *subject
+ 0, // const char *reply_to
+ 0, // const char *content_type
+ 0, // const char *content_encoding
+ 0, // int32_t correlation_id
+ conn->config->site);
}
if (receive_complete) {
qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = true in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
- if (!stream_data->body) {
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_compose_insert_binary(stream_data->body, 0, 0);
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"] Inserting empty body data in compose_and_deliver", conn->conn_id, stream_data->stream_id);
- }
-
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
+
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ stream_data->body_data_added_to_msg = true;
}
+
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
- if (stream_data->body) {
- qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ if (DEQ_SIZE(stream_data->body_buffers) > 0) {
+ qd_log(http2_adaptor->log_source, QD_LOG_TRACE, "[C%"PRIu64"][S%"PRId32"][L%"PRIu64"] receive_complete = false and has stream_data->body_buffers in compose_and_deliver", conn->conn_id, stream_data->stream_id, stream_data->in_link->identity);
+ bool q2_blocked;
if (stream_data->footer_properties) {
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
+ qd_buffer_list_t existing_buffers;
+ DEQ_INIT(existing_buffers);
+ qd_compose_take_buffers(stream_data->footer_properties, &existing_buffers);
+ qd_message_stream_data_footer_append(stream_data->message, &existing_buffers);
}
else {
- qd_message_compose_4(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, &q2_blocked);
}
- stream_data->body_data_added = true;
+ stream_data->body_data_added_to_msg = true;
+ conn->q2_blocked = conn->q2_blocked || q2_blocked;
+ if (conn->q2_blocked) {
+ qd_log(http2_adaptor->protocol_log_source, QD_LOG_TRACE, "[C%"PRIu64"] q2 is blocked on this connection", conn->conn_id);
+ }
}
else {
-
if (stream_data->footer_properties) {
+
+ if (!stream_data->entire_footer_arrived) {
+ qd_compose_free(header_and_props);
+ return false;
+ }
+
//
// The footer has already arrived but there was no body. Insert an empty body
//
- stream_data->body = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_message_compose_5(stream_data->message, header_and_props, stream_data->app_properties, stream_data->body, stream_data->footer_properties, receive_complete);
+ qd_message_compose_3(stream_data->message, header_and_props, stream_data->app_properties, receive_complete);
+ qd_message_stream_data_append(stream_data->message, &stream_data->body_buffers, 0);
Review comment:
Ok. Personally I've been unpleasantly surprised by Q2 behavior far to many times to trust any assumption about what it's up to. And I love asserts(), can't get enough of them.
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org
[GitHub] [qpid-dispatch] kgiusti commented on a change in pull request #1301: DISPATCH-1940, DISPATCH-2198: Added code to handle http2 and q2 flow …
Posted by GitBox <gi...@apache.org>.
kgiusti commented on a change in pull request #1301:
URL: https://github.com/apache/qpid-dispatch/pull/1301#discussion_r675790413
##########
File path: tests/system_tests_http2.py
##########
@@ -728,3 +741,126 @@ def test_goaway(self):
address = self.router_qdra.http_addresses[0] + "/goaway_test_1"
out = self.run_curl(address, args=["-i"])
self.assertIn("HTTP/2 503", out)
+
+
+class Http2Q2OneRouterTest(Http2TestBase):
+ @classmethod
+ def setUpClass(cls):
+ super(Http2Q2OneRouterTest, cls).setUpClass()
+ if skip_h2_test():
+ return
+ cls.http2_server_name = "http2_slow_q2_server"
+ os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port())
+ cls.http2_server = cls.tester.http2server(name=cls.http2_server_name,
+ listen_port=int(os.getenv('SERVER_LISTEN_PORT')),
+ py_string='python3',
+ server_file="http2_slow_q2_server.py")
+ name = "http2-test-router"
+ cls.connector_name = 'connectorToServer'
+ cls.connector_props = {
+ 'port': os.getenv('SERVER_LISTEN_PORT'),
+ 'address': 'examples',
+ 'host': '127.0.0.1',
+ 'protocolVersion': 'HTTP2',
+ 'name': cls.connector_name
+ }
+ config = Qdrouterd.Config([
+ ('router', {'mode': 'standalone', 'id': 'QDR'}),
+ ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}),
+
+ ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples',
+ 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}),
+ ('httpConnector', cls.connector_props)
+ ])
+ cls.router_qdra = cls.tester.qdrouterd(name, config, wait=True)
+
+ @unittest.skipIf(skip_h2_test(),
+ "Python 3.7 or greater, hyper-h2 and curl needed to run hyperhttp2 tests")
+ def test_q2_block_unblock(self):
+ # curl -X POST -H "Content-Type: multipart/form-data" -F "data=@/home/gmurthy/opensource/test.jpg"
+ # http://127.0.0.1:9000/upload --http2-prior-knowledge
+ address = self.router_qdra.http_addresses[0] + "/upload"
+ out = self.run_curl(address, args=['-X', 'POST', '-H', 'Content-Type: multipart/form-data',
+ '-F', 'data=@' + image_file('test.jpg')])
+ self.assertIn('Success', out)
+ num_blocked = 0
+ num_unblocked = 0
+ blocked = "blocked"
+ unblocked = "unblocked"
+ with open(self.router_qdra.logfile_path, 'r') as router_log:
+ log_lines = router_log.read().split("\n")
+ for log_line in log_lines:
+ if unblocked in log_line:
+ num_unblocked += 1
+ elif blocked in log_line:
+ num_blocked += 1
+
+ self.assertGreater(num_blocked, 0)
+ self.assertGreater(num_unblocked, 0)
Review comment:
Understood
--
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: dev-unsubscribe@qpid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org