DEBUG DO NOT MERGE

Author: kgiusti

src/adaptors/tcp_lite/tcp_lite.c

@@ -1261,7 +1261,7 @@ static bool manage_flow_XSIDE_IO(tcplite_connection_t *conn)
         conn->inbound_octets += octet_count;
 
         if (octet_count > 0) {
-            qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] %cSIDE Raw read: Produced %"PRIu64" octets into stream", conn->conn_id, conn->listener_side ? 'L' : 'C', octet_count);
+            qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] !KAG! %cSIDE Raw read: Produced %"PRIu64" octets into stream", conn->conn_id, conn->listener_side ? 'L' : 'C', octet_count);
             if (!was_blocked && window_full(conn)) {
                 conn->window.closed_count += 1;
                 qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP RX window CLOSED: inbound_bytes=%" PRIu64 " unacked=%" PRIu64,
@@ -1324,7 +1324,7 @@ static bool manage_flow_XSIDE_IO(tcplite_connection_t *conn)
             conn->outbound_octets += body_octets;
             conn->window.pending_ack += body_octets;
             if (body_octets > 0) {
-                qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] %cSIDE Raw write: Consumed %"PRIu64" octets from stream (body-field)", conn->conn_id, conn->listener_side ? 'L' : 'C', body_octets);
+                qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] !KAG! %cSIDE Raw write: Consumed %"PRIu64" octets from stream (body-field)", conn->conn_id, conn->listener_side ? 'L' : 'C', body_octets);
             }
         }
 
@@ -1336,7 +1336,7 @@ static bool manage_flow_XSIDE_IO(tcplite_connection_t *conn)
             conn->outbound_octets += octets;
             conn->window.pending_ack += octets;
             if (octets > 0) {
-                qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] %cSIDE Raw write: Consumed %"PRIu64" octets from stream", conn->conn_id, conn->listener_side ? 'L' : 'C', octets);
+                qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] !KAG! %cSIDE Raw write: Consumed %"PRIu64" octets from stream", conn->conn_id, conn->listener_side ? 'L' : 'C', octets);
             }
         }
 

src/message.c

@@ -1570,7 +1570,7 @@ static void qd_message_receive_cutthrough(qd_message_t *in_msg, pn_delivery_t *d
             // Data received, advance the producer slot pointer
             //
             notify_produced = true;
-            qd_log(LOG_MESSAGE, QD_LOG_DEBUG, "qd_message_receive_cutthrough - %u octets written to use_slot=%u",
+            qd_log(LOG_MESSAGE, QD_LOG_DEBUG, "qd_message_receive_cutthrough !KAG! - %u octets written to use_slot=%u",
                    qd_buffer_list_length(&content->uct_slots[use_slot]), use_slot);
             sys_atomic_set(&content->uct_produce_slot, (use_slot + 1) % UCT_SLOT_COUNT);
 
@@ -1887,16 +1887,21 @@ static void qd_message_send_cut_through(qd_message_pvt_t *msg, qd_message_conten
     *q3_stalled = !IS_ATOMIC_FLAG_SET(&content->aborted) && (pn_session_outgoing_bytes(pns) >= q3_upper);
     while (!*q3_stalled && (sys_atomic_get(&content->uct_consume_slot) - sys_atomic_get(&content->uct_produce_slot)) % UCT_SLOT_COUNT != 0) {
         uint32_t use_slot = sys_atomic_get(&content->uct_consume_slot);
+        size_t kag_size = 0;
 
         qd_buffer_t *buf = DEQ_HEAD(content->uct_slots[use_slot]);
         while (!!buf) {
             DEQ_REMOVE_HEAD(content->uct_slots[use_slot]);
             if (!IS_ATOMIC_FLAG_SET(&content->aborted)) {
-                pn_link_send(pnl, (char*) qd_buffer_base(buf), qd_buffer_size(buf));
+                ssize_t sent = pn_link_send(pnl, (char*) qd_buffer_base(buf), qd_buffer_size(buf));
+                (void) sent;
+                assert(sent == qd_buffer_size(buf));
+                kag_size += qd_buffer_size(buf);
             }
             qd_buffer_free(buf);
             buf = DEQ_HEAD(content->uct_slots[use_slot]);
         }
+        qd_log(LOG_MESSAGE, QD_LOG_DEBUG, "qd_message_send_cut_through !KAG! use_slot=%"PRIu32" consumed %zu octets", use_slot, kag_size);
 
         sys_atomic_set(&content->uct_consume_slot, (use_slot + 1) % UCT_SLOT_COUNT);
         notify_consumed = true;
@@ -3313,6 +3318,8 @@ void qd_message_produce_buffers(qd_message_t *stream, qd_buffer_list_t *buffers)
     assert(qd_message_can_produce_buffers(stream));
 
     uint32_t useSlot = sys_atomic_get(&content->uct_produce_slot);
+    qd_log(LOG_MESSAGE, QD_LOG_DEBUG, "qd_message_produce_buffers !KAG! use_slot=%"PRIu32" produced %u octets",
+           useSlot, qd_buffer_list_length(buffers));
     DEQ_MOVE(*buffers, content->uct_slots[useSlot]);
     sys_atomic_set(&content->uct_produce_slot, (useSlot + 1) % UCT_SLOT_COUNT);
     cutthrough_notify_buffers_produced_inbound(stream);
@@ -3332,14 +3339,15 @@ int qd_message_consume_buffers(qd_message_t *stream, qd_buffer_list_t *buffers,
 
     while (count < limit && !empty) {
         uint32_t useSlot = sys_atomic_get(&content->uct_consume_slot);
-        qd_log(LOG_MESSAGE, QD_LOG_DEBUG, "qd_message_consume_buffers useSlot=%"PRIu32"", useSlot);
+        size_t kag_total = 0;
         while (count < limit && !DEQ_IS_EMPTY(content->uct_slots[useSlot])) {
             qd_buffer_t *buf = DEQ_HEAD(content->uct_slots[useSlot]);
-            qd_log(LOG_MESSAGE, QD_LOG_DEBUG, "qd_message_consume_buffers buf size=%zu", qd_buffer_size(buf));
             DEQ_REMOVE_HEAD(content->uct_slots[useSlot]);
             DEQ_INSERT_TAIL(*buffers, buf);
             count++;
+            kag_total += qd_buffer_size(buf);
         }
+        qd_log(LOG_MESSAGE, QD_LOG_DEBUG, "qd_message_consume_buffers !KAG! use_slot=%"PRIu32" consumed %zu octets", useSlot, kag_total);
         if (DEQ_IS_EMPTY(content->uct_slots[useSlot])) {
             notify_consumed = true;
             sys_atomic_set(&content->uct_consume_slot, (useSlot + 1) % UCT_SLOT_COUNT);