Checkpoint: draft. Has locking bug that needs fixin'

Author: kgiusti

include/qpid/dispatch/cutthrough_utils.h

@@ -27,7 +27,7 @@
  * 
  * @param msg - Pointer to a stream
  */
-void cutthrough_notify_buffers_produced_inbound(qd_message_t *msg);
-void cutthrough_notify_buffers_consumed_outbound(qd_message_t *msg);
+void cutthrough_notify_buffers_produced_inbound(const qd_message_activation_t *activation);
+void cutthrough_notify_buffers_consumed_outbound(const qd_message_activation_t *activation);
 
 #endif

include/qpid/dispatch/message.h

@@ -754,20 +754,6 @@ void qd_message_produce_buffers(qd_message_t *stream, qd_buffer_list_t *buffers)
 int qd_message_consume_buffers(qd_message_t *stream, qd_buffer_list_t *buffers, int limit);
 
 
-/**
- * Indicate whether this stream should be resumed from a stalled state.  This will be the case
- * if (a) the stream was stalled due to being full, and (b) the payload has shrunk down below
- * the resume threshold.
- *
- * If the result is true, there is a side effect of clearing the 'stalled' state.
- *
- * @param stream Pointer to the message
- * @return true Yes, the stream was stalled and buffer production may continue
- * @return false No, the stream was not stalled or it was stalled and is not yet ready to resume
- */
-bool qd_message_resume_from_stalled(qd_message_t *stream);
-
-
 typedef enum {
     QD_ACTIVATION_NONE = 0,
     QD_ACTIVATION_AMQP,
@@ -784,33 +770,31 @@ typedef struct {
  * Tell the message stream which connection is consuming its buffers.
  *
  * @param stream Pointer to the message
- * @param connection Pointer to the qd_connection that is consuming this stream's buffers
+ * @param activation Parameters for activating the consuming I/O thread
  */
 void qd_message_set_consumer_activation(qd_message_t *stream, qd_message_activation_t *activation);
 
 /**
- * Return the connection that is consuming this message stream's buffers.
+ * Cancel the activation. No further activations will be occur on return from this call.
  *
  * @param stream Pointer to the message
- * @return qd_connection_t* Pointer to the connection that is consuming buffers from this stream
  */
-void qd_message_get_consumer_activation(const qd_message_t *stream, qd_message_activation_t *activation);
+void qd_message_cancel_consumer_activation(qd_message_t *stream);
 
 /**
  * Tell the message stream which connection is producing its buffers.
  *
  * @param stream Pointer to the message
- * @param connection Pointer to the qd_connection that is consuming this stream's buffers
+ * @param activation Parameters for activating the producing I/O thread
  */
 void qd_message_set_producer_activation(qd_message_t *stream, qd_message_activation_t *activation);
 
 /**
- * Return the connection that is producing this message stream's buffers.
+ * Cancel the activation. No further activations will occur on return from this call.
  *
  * @param stream Pointer to the message
- * @return qd_connection_t* Pointer to the connection that is consuming buffers from this stream
  */
-void qd_message_get_producer_activation(const qd_message_t *stream, qd_message_activation_t *activation);
+void qd_message_cancel_producer_activation(qd_message_t *stream);
 
 ///@}
 

src/adaptors/tcp_lite/tcp_lite.c

@@ -515,17 +515,14 @@ static void close_connection_XSIDE_IO(tcplite_connection_t *conn, bool no_delay)
 
     free(conn->reply_to);
 
-    qd_message_activation_t activation;
-    activation.type     = QD_ACTIVATION_NONE;
-    activation.delivery = 0;
-    qd_nullify_safe_ptr(&activation.safeptr);
-
     if (!!conn->inbound_stream) {
-        qd_message_set_producer_activation(conn->inbound_stream, &activation);
+        qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP cancel producer activation", DLV_ARGS(conn->inbound_delivery));
+        qd_message_cancel_producer_activation(conn->inbound_stream);
     }
 
     if (!!conn->outbound_stream) {
-        qd_message_set_consumer_activation(conn->outbound_stream, &activation);
+        qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP cancel consuer activation", DLV_ARGS(conn->outbound_delivery));
+        qd_message_cancel_consumer_activation(conn->outbound_stream);
     }
 
     if (!!conn->inbound_delivery) {
@@ -890,6 +887,17 @@ static void link_setup_CSIDE_IO(tcplite_connection_t *conn, qdr_delivery_t *deli
     qdr_link_set_context(conn->inbound_link, conn);
     conn->outbound_link = qdr_link_first_attach(conn->core_conn, QD_OUTGOING, qdr_terminus(0), qdr_terminus(0), "tcp.cside.out", 0, false, delivery, &conn->outbound_link_id);
     qdr_link_set_context(conn->outbound_link, conn);
+
+    // now that the raw connection is up and able to be activated enable cutthrough activation
+
+    assert(conn->outbound_stream);
+    qd_message_activation_t activation;
+    activation.type     = QD_ACTIVATION_TCP;
+    activation.delivery = 0;
+    qd_alloc_set_safe_ptr(&activation.safeptr, conn);
+    qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP enabling consumer activation", DLV_ARGS(delivery));
+    qd_message_set_consumer_activation(conn->outbound_stream, &activation);
+    qd_message_start_unicast_cutthrough(conn->outbound_stream);
 }
 
 
@@ -956,6 +964,7 @@ static bool try_compose_and_send_client_stream_LSIDE_IO(tcplite_connection_t *co
     activation.type     = QD_ACTIVATION_TCP;
     activation.delivery = 0;
     qd_alloc_set_safe_ptr(&activation.safeptr, conn);
+    qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%" PRIu64 "][L%" PRIu64 "] TCP enabling producer activation", conn->conn_id, conn->inbound_link_id);
     qd_message_set_producer_activation(conn->inbound_stream, &activation);
     qd_message_start_unicast_cutthrough(conn->inbound_stream);
 
@@ -1022,6 +1031,7 @@ static void compose_and_send_server_stream_CSIDE_IO(tcplite_connection_t *conn)
     activation.type     = QD_ACTIVATION_TCP;
     activation.delivery = 0;
     qd_alloc_set_safe_ptr(&activation.safeptr, conn);
+    qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%" PRIu64 "][L%" PRIu64 "] TCP enabling producer activation", conn->conn_id, conn->inbound_link_id);
     qd_message_set_producer_activation(conn->inbound_stream, &activation);
     qd_message_start_unicast_cutthrough(conn->inbound_stream);
 
@@ -1092,6 +1102,7 @@ static uint64_t handle_outbound_delivery_LSIDE_IO(tcplite_connection_t *conn, qd
         activation.type     = QD_ACTIVATION_TCP;
         activation.delivery = 0;
         qd_alloc_set_safe_ptr(&activation.safeptr, conn);
+        qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP enabling consumer activation", DLV_ARGS(delivery));
         qd_message_set_consumer_activation(conn->outbound_stream, &activation);
         qd_message_start_unicast_cutthrough(conn->outbound_stream);
     }
@@ -1196,16 +1207,6 @@ static uint64_t handle_first_outbound_delivery_CSIDE(tcplite_connector_t *cr, qd
     conn->raw_conn = pn_raw_connection();
     pn_raw_connection_set_context(conn->raw_conn, &conn->context);
 
-    //
-    // Message validation ensures the start of the message body is present. Activate cutthrough on the body data.
-    //
-    qd_message_activation_t activation;
-    activation.type     = QD_ACTIVATION_TCP;
-    activation.delivery = 0;
-    qd_alloc_set_safe_ptr(&activation.safeptr, conn);
-    qd_message_set_consumer_activation(conn->outbound_stream, &activation);
-    qd_message_start_unicast_cutthrough(conn->outbound_stream);
-
     //
     // The raw connection establishment must be the last thing done in this function.
     // After this call, a separate IO thread may immediately be invoked in the context
@@ -1289,9 +1290,11 @@ static bool manage_flow_XSIDE_IO(tcplite_connection_t *conn)
         if (pn_raw_connection_is_read_closed(conn->raw_conn) && !blocked) {
             qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] Read-closed - close inbound delivery", conn->conn_id);
             qd_message_set_receive_complete(conn->inbound_stream);
+            qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP cancel producer activation", DLV_ARGS(conn->inbound_delivery));
+            qd_message_cancel_producer_activation(conn->inbound_stream);
             qdr_delivery_continue(tcplite_context->core, conn->inbound_delivery, false);
             qdr_delivery_set_context(conn->inbound_delivery, 0);
-            qdr_delivery_decref(tcplite_context->core, conn->inbound_delivery, "TCP_LSIDE_IO - read-close");
+            qdr_delivery_decref(tcplite_context->core, conn->inbound_delivery, "FLOW_XSIDE_IO - inbound_delivery released");
             conn->inbound_delivery = 0;
             conn->inbound_stream   = 0;
             return true;
@@ -1360,8 +1363,12 @@ static bool manage_flow_XSIDE_IO(tcplite_connection_t *conn)
         if (qd_message_receive_complete(conn->outbound_stream) && !qd_message_can_consume_buffers(conn->outbound_stream)) {
             qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] Rx-complete, rings empty: Write-closing the raw connection", conn->conn_id);
             pn_raw_connection_write_close(conn->raw_conn);
+            qd_message_set_send_complete(conn->outbound_stream);
+            qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP cancel consumer activation", DLV_ARGS(conn->outbound_delivery));
+            qd_message_cancel_consumer_activation(conn->outbound_stream);
             qdr_delivery_set_context(conn->outbound_delivery, 0);
             qdr_delivery_remote_state_updated(tcplite_context->core, conn->outbound_delivery, PN_ACCEPTED, true, 0, true); // accepted, settled, ref_given
+            // do NOT decref outbound_delivery - ref count passed to qdr_delivery_remote_state_updated()!
             conn->outbound_delivery = 0;
             conn->outbound_stream   = 0;
         } else {
@@ -1532,15 +1539,11 @@ static bool manage_tls_flow_XSIDE_IO(tcplite_connection_t *conn)
     //
     bool ignore;
     if (qd_tls_is_input_drained(conn->tls, &ignore) && conn->inbound_stream) {
-        qd_message_activation_t activation;
-        activation.type     = QD_ACTIVATION_NONE;
-        activation.delivery = 0;
-        qd_nullify_safe_ptr(&activation.safeptr);
-
         qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TLS inbound stream receive complete", DLV_ARGS(conn->inbound_delivery));
 
         qd_message_set_receive_complete(conn->inbound_stream);
-        qd_message_set_producer_activation(conn->inbound_stream, &activation);
+        qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP cancel producer activation", DLV_ARGS(conn->inbound_delivery));
+        qd_message_cancel_producer_activation(conn->inbound_stream);
 
         qdr_delivery_set_context(conn->inbound_delivery, 0);
         qdr_delivery_continue(tcplite_context->core, conn->inbound_delivery, false);
@@ -1553,21 +1556,17 @@ static bool manage_tls_flow_XSIDE_IO(tcplite_connection_t *conn)
     // Check for end of outbound message data
     //
     if (qd_tls_is_output_flushed(conn->tls) && conn->outbound_stream) {
-        qd_message_activation_t activation;
-        activation.type     = QD_ACTIVATION_NONE;
-        activation.delivery = 0;
-        qd_nullify_safe_ptr(&activation.safeptr);
-
         qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TLS outbound stream send complete", DLV_ARGS(conn->outbound_delivery));
 
         qd_message_set_send_complete(conn->outbound_stream);
-        qd_message_set_consumer_activation(conn->outbound_stream, &activation);
+        qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP cancel consumer activation", DLV_ARGS(conn->outbound_delivery));
+        qd_message_cancel_consumer_activation(conn->outbound_stream);
 
         qdr_delivery_set_context(conn->outbound_delivery, 0);
         qdr_delivery_remote_state_updated(tcplite_context->core, conn->outbound_delivery,
                                           tls_status < 0 ? PN_MODIFIED : PN_ACCEPTED,
-                                          true, 0, false); // settled, 0, ref_given
-        qdr_delivery_decref(tcplite_context->core, conn->outbound_delivery, "TLS_FLOW_XSIDE_IO - outbound_delivery released");
+                                          true, 0, true); // settled, 0, ref_given
+        // do NOT decref outbound_delivery - ref count passed to qdr_delivery_remote_state_updated()!
         conn->outbound_delivery = 0;
         conn->outbound_stream   = 0;
     }

src/cutthrough_utils.c

@@ -25,7 +25,7 @@
 #include "adaptors/tcp_lite/tcp_lite.h"
 
 
-static void activate_connection(qd_message_activation_t *activation, qd_direction_t dir)
+static void activate_connection(const qd_message_activation_t *activation, qd_direction_t dir)
 {
     switch (activation->type) {
     case QD_ACTIVATION_NONE:
@@ -77,20 +77,13 @@ static void activate_connection(qd_message_activation_t *activation, qd_directio
 }
 
 
-void cutthrough_notify_buffers_produced_inbound(qd_message_t *msg)
+void cutthrough_notify_buffers_produced_inbound(const qd_message_activation_t *activation)
 {
-    qd_message_activation_t activation;
-    qd_message_get_consumer_activation(msg, &activation);
-    activate_connection(&activation, QD_OUTGOING);
+    activate_connection(activation, QD_OUTGOING);
 }
 
 
-void cutthrough_notify_buffers_consumed_outbound(qd_message_t *msg)
+void cutthrough_notify_buffers_consumed_outbound(const qd_message_activation_t *activation)
 {
-    bool unstall = qd_message_resume_from_stalled(msg);
-    if (unstall) {
-        qd_message_activation_t activation;
-        qd_message_get_producer_activation(msg, &activation);
-        activate_connection(&activation, QD_INCOMING);
-    }
+    activate_connection(activation, QD_INCOMING);
 }

src/message.c

@@ -1528,6 +1528,31 @@ bool qd_message_has_data_in_content_or_pending_buffers(qd_message_t   *msg)
     return false;
 }
 
+static inline void activate_message_consumer(qd_message_t *stream)
+{
+    qd_message_content_t *content = MSG_CONTENT(stream);
+    LOCK(&content->lock);
+    if (content->uct_consumer_activation.type != QD_ACTIVATION_NONE) {
+        cutthrough_notify_buffers_produced_inbound(&content->uct_consumer_activation);
+    }
+    UNLOCK(&content->lock);
+}
+
+
+static inline void activate_message_producer(qd_message_t *stream)
+{
+    qd_message_content_t *content = MSG_CONTENT(stream);
+
+    uint32_t full_slots = (sys_atomic_get(&content->uct_produce_slot) - sys_atomic_get(&content->uct_consume_slot)) % UCT_SLOT_COUNT;
+    if (full_slots < UCT_RESUME_THRESHOLD) {
+        LOCK(&content->lock);
+        if (content->uct_producer_activation.type != QD_ACTIVATION_NONE) {
+            cutthrough_notify_buffers_consumed_outbound(&content->uct_producer_activation);
+        }
+        UNLOCK(&content->lock);
+    }
+}
+
 
 // Read incoming data from a pn_link_t and store it into a buffer list. Limit buffer list length to a maximum of
 // limit. Helper routine for qd_message_receive_cutthrough()
@@ -1576,7 +1601,6 @@ static void qd_message_receive_cutthrough(qd_message_t *in_msg, pn_delivery_t *d
 
             if ((sys_atomic_get(&content->uct_consume_slot) - sys_atomic_get(&content->uct_produce_slot)) % UCT_SLOT_COUNT == 1) {
                 stalled = true;
-                SET_ATOMIC_FLAG(&content->uct_producer_stalled);
             }
         }
 
@@ -1595,7 +1619,7 @@ static void qd_message_receive_cutthrough(qd_message_t *in_msg, pn_delivery_t *d
     }
 
     if (notify_produced) {
-        cutthrough_notify_buffers_produced_inbound(in_msg);
+        activate_message_consumer(in_msg);
     }
 }
 
@@ -1912,10 +1936,11 @@ static void qd_message_send_cut_through(qd_message_pvt_t *msg, qd_message_conten
         // all of the buffered content.  Mark the message as send-complete.
         //
         SET_ATOMIC_FLAG(&msg->send_complete);
+        notify_consumed = false;  // no need to restart producer - it is done
     }
 
     if (notify_consumed) {
-        cutthrough_notify_buffers_consumed_outbound((qd_message_t *) msg);
+        activate_message_producer((qd_message_t *) msg);
     }
 }
 
@@ -3317,11 +3342,7 @@ void qd_message_produce_buffers(qd_message_t *stream, qd_buffer_list_t *buffers)
     uint32_t useSlot = sys_atomic_get(&content->uct_produce_slot);
     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);
-
-    if ((sys_atomic_get(&content->uct_consume_slot) - sys_atomic_get(&content->uct_produce_slot)) % UCT_SLOT_COUNT == 1) {
-        SET_ATOMIC_FLAG(&content->uct_producer_stalled);
-    }
+    activate_message_consumer(stream);
 }
 
 
@@ -3348,50 +3369,41 @@ int qd_message_consume_buffers(qd_message_t *stream, qd_buffer_list_t *buffers,
     }
 
     if (notify_consumed) {
-        cutthrough_notify_buffers_consumed_outbound(stream);
+        activate_message_producer(stream);
     }
 
     return count;
 }
 
-
-bool qd_message_resume_from_stalled(qd_message_t *stream)
-{
-    qd_message_content_t *content = MSG_CONTENT(stream);
-
-    if (IS_ATOMIC_FLAG_SET(&content->uct_producer_stalled)
-        && (sys_atomic_get(&content->uct_produce_slot) - sys_atomic_get(&content->uct_consume_slot)) % UCT_SLOT_COUNT < UCT_RESUME_THRESHOLD) {
-        CLEAR_ATOMIC_FLAG(&content->uct_producer_stalled);
-        return true;
-    }
-
-    return false;
-}
-
-
 void qd_message_set_consumer_activation(qd_message_t *stream, qd_message_activation_t *activation)
 {
     qd_message_content_t *content = MSG_CONTENT(stream);
+    LOCK(&content->lock);
     content->uct_consumer_activation = *activation;
+    UNLOCK(&content->lock);
 }
 
 
-void qd_message_get_consumer_activation(const qd_message_t *stream, qd_message_activation_t *activation)
+void qd_message_cancel_consumer_activation(qd_message_t *stream)
 {
     qd_message_content_t *content = MSG_CONTENT(stream);
-    *activation = content->uct_consumer_activation;
+    LOCK(&content->lock);
+    content->uct_consumer_activation.type = QD_ACTIVATION_NONE;
+    UNLOCK(&content->lock);
 }
 
-
 void qd_message_set_producer_activation(qd_message_t *stream, qd_message_activation_t *activation)
 {
     qd_message_content_t *content = MSG_CONTENT(stream);
+    LOCK(&content->lock);
     content->uct_producer_activation = *activation;
+    UNLOCK(&content->lock);
 }
 
-
-void qd_message_get_producer_activation(const qd_message_t *stream, qd_message_activation_t *activation)
+void qd_message_cancel_producer_activation(qd_message_t *stream)
 {
     qd_message_content_t *content = MSG_CONTENT(stream);
-    *activation = content->uct_producer_activation;
+    LOCK(&content->lock);
+    content->uct_producer_activation.type = QD_ACTIVATION_NONE;
+    UNLOCK(&content->lock);
 }