fix 3243

Fix IOBuf TLS block pool self-loop caused by double-return (#3243)

release_tls_block() and release_tls_block_chain() do not guard against
a block being returned to TLS when it is already the list head.  The
assignment `b->portal_next = block_head` becomes `b->portal_next = b`,
forming a self-loop that causes remove_tls_block_chain() or
share_tls_block() to spin forever, silently hanging the thread at exit.

Fix:
- release_tls_block(): add early return when b == block_head, skipping
  the duplicate insertion.
- release_tls_block_chain(): during the existing chain walk, check each
  node against block_head before linking.  Return early if overlap is
  detected so that num_blocks stays consistent with the actual list
  length (remove_tls_block_chain verifies this via CHECK_EQ).

Add three unit tests that reproduce the self-loop through:
1. Direct double release_tls_block() of the same block.
2. release_tls_block_chain() with a chain overlapping the TLS head.
3. IOBufAsZeroCopyOutputStream::BackUp() followed by a second release.

Author: walterzhaoJR

src/butil/iobuf.cpp

@@ -323,31 +323,47 @@ IOBuf::Block* share_tls_block() {
 // NOTE: b MUST be non-NULL and all blocks linked SHOULD not be full.
 void release_tls_block_chain(IOBuf::Block* b) {
     TLSData& tls_data = g_tls_data;
-    size_t n = 0;
+    size_t n_hit_threshold = 0;
     if (tls_data.num_blocks >= max_blocks_per_thread()) {
         do {
-            ++n;
             IOBuf::Block* const saved_next = b->u.portal_next;
-            b->dec_ref();
+            // If b is already cached in TLS, dec_ref() would drop a reference
+            // still owned by the TLS list and leave a dangling pointer behind.
+            if (!is_in_tls_block_chain(tls_data.block_head, b)) {
+                b->dec_ref();
+                ++n_hit_threshold;
+            }
             b = saved_next;
         } while (b);
-        g_num_hit_tls_threshold.fetch_add(n, butil::memory_order_relaxed);
+        g_num_hit_tls_threshold.fetch_add(
+            n_hit_threshold, butil::memory_order_relaxed);
         return;
     }
     IOBuf::Block* first_b = b;
     IOBuf::Block* last_b = NULL;
+    size_t n_to_tls = 0;
     do {
-        ++n;
         CHECK(!b->full());
+        // Guard against overlap with any node already cached in TLS, not just
+        // block_head.  Returning X into H -> X would create a 2-node cycle
+        // X -> H -> X.  If the overlap happens after a unique prefix such as
+        // A -> X, keep the prefix by linking A directly to the old TLS head.
+        if (is_in_tls_block_chain(tls_data.block_head, b)) {
+            break;
+        }
+        ++n_to_tls;
+        last_b = b;
         if (b->u.portal_next == NULL) {
-            last_b = b;
             break;
         }
         b = b->u.portal_next;
     } while (true);
+    if (last_b == NULL) {
+        return;
+    }
     last_b->u.portal_next = tls_data.block_head;
     tls_data.block_head = first_b;
-    tls_data.num_blocks += n;
+    tls_data.num_blocks += n_to_tls;
     if (!tls_data.registered) {
         tls_data.registered = true;
         butil::thread_atexit(remove_tls_block_chain);

src/butil/iobuf_inl.h

@@ -603,12 +603,28 @@ void remove_tls_block_chain();
 
 IOBuf::Block* acquire_tls_block();
 
+static inline bool is_in_tls_block_chain(IOBuf::Block* head, IOBuf::Block* b) {
+    for (IOBuf::Block* p = head; p != NULL; p = p->u.portal_next) {
+        if (p == b) {
+            return true;
+        }
+    }
+    return false;
+}
+
 // Return one block to TLS.
 inline void release_tls_block(IOBuf::Block* b) {
     if (!b) {
         return;
     }
     TLSData *tls_data = get_g_tls_data();
+    // Guard against duplicate return anywhere in the TLS list.  Checking only
+    // block_head misses cases like H -> X where returning X again would create
+    // a 2-node cycle X -> H -> X.  The TLS list is short (soft-limited), so a
+    // linear scan is cheap here.
+    if (is_in_tls_block_chain(tls_data->block_head, b)) {
+        return;
+    }
     if (b->full()) {
         b->dec_ref();
     } else if (tls_data->num_blocks >= max_blocks_per_thread()) {