Move the decrement of the content ref-count below the buffer-trimming logic in qd_message_free.  This eliminates the case where message content can be freed by one thread while another is doing buffer-trimming on the same content.
@@ -927,8 +927,29 @@ void qd_message_free(qd_message_t *in_msg)
qd_message_content_t *content = msg->content;
- rc = sys_atomic_dec(&content->ref_count) - 1;
+ if (msg->is_fanout) {
+ //
+ // Adjust the content's fanout count and decrement all buffer fanout
+ // counts starting with the msg cursor. If the buffer count drops to
+ // zero we can free it.
+ //
+ LOCK(content->lock);
+
+ qd_buffer_t *buf = msg->cursor.buffer;
+ while (buf) {
+ qd_buffer_t *next_buf = DEQ_NEXT(buf);
+ if (qd_buffer_dec_fanout(buf) == 1) {
+ DEQ_REMOVE(content->buffers, buf);
+ qd_buffer_free(buf);
+ }
+ buf = next_buf;
+ }
+ --content->fanout;
+
+ UNLOCK(content->lock);
+ }
+ rc = sys_atomic_dec(&content->ref_count) - 1;
if (rc == 0) {
if (content->ma_field_iter_in)
qd_iterator_free(content->ma_field_iter_in);
@@ -948,28 +969,8 @@ void qd_message_free(qd_message_t *in_msg)
sys_mutex_free(content->lock);
free_qd_message_content_t(content);
-
- } else if (msg->is_fanout) {
- //
- // Adjust the content's fanout count and decrement all buffer fanout
- // counts starting with the msg cursor. If the buffer count drops to
- // zero we can free it.
- //
- LOCK(content->lock);
-
- qd_buffer_t *buf = msg->cursor.buffer;
- while (buf) {
- qd_buffer_t *next_buf = DEQ_NEXT(buf);
- if (qd_buffer_dec_fanout(buf) == 1) {
- DEQ_REMOVE(content->buffers, buf);
- qd_buffer_free(buf);
}
- buf = next_buf;
- }
- --content->fanout;
- UNLOCK(content->lock);
- }
free_qd_message_t((qd_message_t*) msg);
}
