malloc: Implement deep consolidation during merging

Backwards consolidation happens unconditionally during batch
processing.  Forward consolidation does not because of the
need to skip over a chunk to get its metadata, and that is not
possible if the chunk may be freed concurrently.

malloc/malloc.c

@@ -3007,41 +3007,166 @@ tcache_free_init (void *mem)
   __libc_free (mem);
 }
 
-/* If the arena does not change between chunks, keep the lock.  */
-static inline void
-__libc_free_batched_loop (bool do_lock, mstate av, mchunkptr p, INTERNAL_SIZE_T size,
+#if USE_TCACHE
+static __always_inline mstate
+__libc_free_batched_lock (bool do_lock, mstate av, mchunkptr p)
+{
+  if (do_lock && av == NULL)
+    {
+      av = arena_for_chunk (p);
+      __libc_lock_lock (av->mutex);
+    }
+  return av;
+}
+
+/* Try consolidating backwards.  *AV is locked opportunistically if
+   consolidation appears possible.  *EXPECTED_BEFORE is adjusted
+   accordingly.  */
+static __always_inline void
+__libc_free_batched_consolidate_backwards (bool do_lock, mstate *av,
+					   void **expected_before,
+					   mchunkptr p, INTERNAL_SIZE_T size)
+{
+  /* See the consolidation code in _int_free_merge_chunk.  */
+  if (!prev_inuse (p))
+    {
+      *av = __libc_free_batched_lock (do_lock, *av, p);
+      /* Recheck under the arena lock.  */
+      if (!prev_inuse (p))
+	{
+	  INTERNAL_SIZE_T prevsize = prev_size (p);
+	  p = chunk_at_offset(p, -((long) prevsize));
+	  if (__glibc_unlikely (chunksize(p) != prevsize))
+	    malloc_printerr ("corrupted size vs. prev_size"
+			     " while batch consolidating");
+	  unlink_chunk (*av, p);
+	  *expected_before = (void *) p - size;
+	}
+    }
+}
+
+/* Try consolidating forward.  *EXPECTED_AFTER is updated accordingly.  */
+static __always_inline void
+__libc_free_batched_consolidate_forward (bool do_lock, mstate av,
+					 void **expected_after,
+					 mchunkptr p, INTERNAL_SIZE_T size)
+{
+  if (do_lock && av == NULL)
+    /* It is not possible to check the in-use status of the next chunk
+       without the arena lock.  Its chunk size may change and move the
+       location of the in-use metadata bit.  */
+    return;
+  /* See the start of _int_free_create_chunk.  */
+  mchunkptr nextchunk = chunk_at_offset (p, size);
+  if (nextchunk == av->top)
+    return;
+  INTERNAL_SIZE_T nextsize = chunksize (nextchunk);
+  if (!inuse_bit_at_offset (nextchunk, nextsize))
+    {
+      unlink_chunk (av, nextchunk);
+      /* Set the in-use bit because _int_free_merge_chunk checks it.  */
+      set_inuse_bit_at_offset (nextchunk, nextsize);
+      *expected_after = (void *) nextchunk + nextsize;
+    }
+}
+
+/* Free the chunk at EXPECTED_BEFORE + SIZE.  Before that, if AV is
+   NULL, obtain the arena from the chunk and lock it.  Otherwise,
+   assume that AV matches the chunk.  Return the AV value.  */
+static __always_inline mstate
+__libc_free_batched_do_free (bool do_lock, mstate av, void *expected_before,
+			     void *expected_after, INTERNAL_SIZE_T size)
+{
+  mchunkptr chunk = expected_before + size;
+  av = __libc_free_batched_lock (do_lock, av, chunk);
+  _int_free_merge_chunk (av, chunk, expected_after - (void *) chunk);
+  return av;
+}
+
+/* Free part of a tcache chain into the lower-level allocator.  */
+static __always_inline void
+__libc_free_batched_loop (bool do_lock, mchunkptr p, INTERNAL_SIZE_T size,
 			  tcache_perthread_struct *tc, size_t tc_idx)
 {
   /* Empty half of the tcache, for a hysteresis effect.  */
   unsigned int to_free = mp_.tcache_count / 2;
 
+  /* The arena lock is acquired lazily before changes are made to the
+     heap structure.  Merging of adjacent tcache chunks does not need
+     the lock.  */
+  mstate av;
   if (do_lock)
-    __libc_lock_lock (av->mutex);
+    av = NULL;
+  else
+    av = &main_arena;
 
-  _int_free_merge_chunk (av, p, size);
+  /* To avoid repeated linking and unlink of bins (or cluttering the
+     unsorted bin) as tcache chunks are freed, adjacent chunks are
+     merged at the tcache level.  A larger chunk is created and freed
+     directly.
+
+     These variables track the expected merge locations of the current
+     chunk that has not yet been committed to the lower-level
+     allocator.  Start with the merge locations of p.  */
+  void *expected_before = (void *) p - size;
+  __libc_free_batched_consolidate_backwards (do_lock, &av,
+					     &expected_before, p, size);
+  void *expected_after = (void *) p + size;
+  __libc_free_batched_consolidate_forward (do_lock, av,
+					   &expected_after, p, size);
 
   while (tc->entries[tc_idx] != NULL && to_free > 0)
     {
       void *mem = tcache_get_n (tc, tc_idx, &tc->entries[tc_idx], false);
       p = mem2chunk (mem);
-      size = chunksize (p);
+      if (size != chunksize (p))
+	malloc_printerr ("free(): corrupted tcache size");
 
-      /* Lock a different arena if necessary.  */
-      if (do_lock)
+      /* Check if the chunk can be merged.  This does not require the
+	 lock because there is no structural change of the heap yet. */
+      if (p == expected_before)
 	{
-	  mstate chunk_av = arena_for_chunk (p);
-	  if (chunk_av != av)
+	  expected_before -= size;
+	  __libc_free_batched_consolidate_backwards (do_lock, &av,
+						     &expected_before, p, size);
+	}
+      else if (p == expected_after)
+	{
+	  expected_after += size;
+	  __libc_free_batched_consolidate_forward (do_lock, av,
+						   &expected_after, p, size);
+	}
+      else
+	{
+	  /* Deallocate the previous chunk that could not be merged.  */
+	  av = __libc_free_batched_do_free (do_lock, av, expected_before,
+					    expected_after, size);
+
+	  /* If locking and the arena changes, release the lock now.
+	     It will be reacquired once there is no more tcache-level
+	     merging.  */
+	  if (do_lock && arena_for_chunk (p) != av)
 	    {
 	      __libc_lock_unlock (av->mutex);
-	      av = chunk_av;
-	      __libc_lock_lock (av->mutex);
+	      av = NULL;
 	    }
+
+	  /* Continue processing with the chunk that came from tcache.  */
+	  expected_before = (void *) p - size;
+	  __libc_free_batched_consolidate_backwards (do_lock, &av,
+						     &expected_before, p, size);
+	  expected_after = (void *) p + size;
+	  __libc_free_batched_consolidate_forward (do_lock, av,
+						   &expected_after, p, size);
 	}
 
-      _int_free_merge_chunk (av, p, size);
       to_free--;
     }
 
+  /* Free the last chunk.  */
+  av = __libc_free_batched_do_free (do_lock, av, expected_before,
+				    expected_after, size);
+
   if (do_lock)
     __libc_lock_unlock (av->mutex);
 }
@@ -3058,10 +3183,11 @@ __libc_free_batched (mchunkptr p, INTERNAL_SIZE_T size,
     return malloc_printerr_tail ("free(): invalid size (batch)");
 
   if (SINGLE_THREAD_P)
-    __libc_free_batched_loop (false, &main_arena, p, size, tc, tc_idx);
+    __libc_free_batched_loop (false, p, size, tc, tc_idx);
   else
-    __libc_free_batched_loop (true, arena_for_chunk (p), p, size, tc, tc_idx);
+    __libc_free_batched_loop (true, p, size, tc, tc_idx);
 }
+#endif
 
 void
 __libc_free (void *mem)
@@ -4055,7 +4181,7 @@ _int_free_merge_chunk (mstate av, mchunkptr p, INTERNAL_SIZE_T size)
 
   free_perturb (chunk2mem(p), size - CHUNK_HDR_SZ);
 
-  /* Consolidate backward.  */
+  /* Consolidate backward.  See __libc_free_batched_consolidate_backwards.  */
   if (!prev_inuse(p))
     {
       INTERNAL_SIZE_T prevsize = prev_size (p);
@@ -4081,6 +4207,8 @@ static INTERNAL_SIZE_T
 _int_free_create_chunk (mstate av, mchunkptr p, INTERNAL_SIZE_T size,
 			mchunkptr nextchunk, INTERNAL_SIZE_T nextsize)
 {
+  /* The start of this function (forward consolidation) is duplicated
+     in __libc_free_batched_consolidate_forward.  */
   if (nextchunk != av->top)
     {
       /* get and clear inuse bit */