8213352: Separate BufferNode allocation from PtrQueueSet

Move free-list management and allocation to new class.

Reviewed-by: tschatzl, sjohanss

src/hotspot/share/gc/g1/dirtyCardQueue.cpp

@@ -147,18 +147,15 @@ uint DirtyCardQueueSet::num_par_ids() {
 }
 
 void DirtyCardQueueSet::initialize(Monitor* cbl_mon,
-                                   Mutex* fl_lock,
+                                   BufferNode::Allocator* allocator,
                                    int process_completed_threshold,
                                    int max_completed_queue,
                                    Mutex* lock,
-                                   DirtyCardQueueSet* fl_owner,
                                    bool init_free_ids) {
   PtrQueueSet::initialize(cbl_mon,
-                          fl_lock,
+                          allocator,
                           process_completed_threshold,
-                          max_completed_queue,
+                          max_completed_queue);
-                          fl_owner);
-  set_buffer_size(G1UpdateBufferSize);
   _shared_dirty_card_queue.set_lock(lock);
   if (init_free_ids) {
     _free_ids = new FreeIdSet(num_par_ids(), _cbl_mon);

src/hotspot/share/gc/g1/dirtyCardQueue.hpp

@@ -118,11 +118,10 @@ public:
   DirtyCardQueueSet(bool notify_when_complete = true);
 
   void initialize(Monitor* cbl_mon,
-                  Mutex* fl_lock,
+                  BufferNode::Allocator* allocator,
                   int process_completed_threshold,
                   int max_completed_queue,
                   Mutex* lock,
-                  DirtyCardQueueSet* fl_owner,
                   bool init_free_ids = false);
 
   // The number of parallel ids that can be claimed to allow collector or

src/hotspot/share/gc/g1/g1BarrierSet.cpp

@@ -55,6 +55,8 @@ G1BarrierSet::G1BarrierSet(G1CardTable* card_table) :
                       make_barrier_set_c2<G1BarrierSetC2>(),
                       card_table,
                       BarrierSet::FakeRtti(BarrierSet::G1BarrierSet)),
+  _satb_mark_queue_buffer_allocator(G1SATBBufferSize, SATB_Q_FL_lock),
+  _dirty_card_queue_buffer_allocator(G1UpdateBufferSize, DirtyCardQ_FL_lock),
   _satb_mark_queue_set(),
   _dirty_card_queue_set()
 {}
@@ -202,3 +204,11 @@ void G1BarrierSet::on_thread_detach(JavaThread* thread) {
   G1ThreadLocalData::satb_mark_queue(thread).flush();
   G1ThreadLocalData::dirty_card_queue(thread).flush();
 }
+
+BufferNode::Allocator& G1BarrierSet::satb_mark_queue_buffer_allocator() {
+  return _satb_mark_queue_buffer_allocator;
+}
+
+BufferNode::Allocator& G1BarrierSet::dirty_card_queue_buffer_allocator() {
+  return _dirty_card_queue_buffer_allocator;
+}

src/hotspot/share/gc/g1/g1BarrierSet.hpp

@@ -39,6 +39,8 @@ class G1CardTable;
 class G1BarrierSet: public CardTableBarrierSet {
   friend class VMStructs;
  private:
+  BufferNode::Allocator _satb_mark_queue_buffer_allocator;
+  BufferNode::Allocator _dirty_card_queue_buffer_allocator;
   G1SATBMarkQueueSet _satb_mark_queue_set;
   DirtyCardQueueSet _dirty_card_queue_set;
 
@@ -79,6 +81,9 @@ class G1BarrierSet: public CardTableBarrierSet {
   virtual void on_thread_attach(JavaThread* thread);
   virtual void on_thread_detach(JavaThread* thread);
 
+  BufferNode::Allocator& satb_mark_queue_buffer_allocator();
+  BufferNode::Allocator& dirty_card_queue_buffer_allocator();
+
   static G1SATBMarkQueueSet& satb_mark_queue_set() {
     return g1_barrier_set()->_satb_mark_queue_set;
   }

src/hotspot/share/gc/g1/g1CollectedHeap.cpp

@@ -1653,6 +1653,28 @@ jint G1CollectedHeap::initialize() {
   BarrierSet::set_barrier_set(bs);
   _card_table = ct;
 
+  G1BarrierSet::satb_mark_queue_set().initialize(this,
+                                                 SATB_Q_CBL_mon,
+                                                 &bs->satb_mark_queue_buffer_allocator(),
+                                                 G1SATBProcessCompletedThreshold,
+                                                 G1SATBBufferEnqueueingThresholdPercent,
+                                                 Shared_SATB_Q_lock);
+
+  // process_completed_threshold and max_completed_queue are updated
+  // later, based on the concurrent refinement object.
+  G1BarrierSet::dirty_card_queue_set().initialize(DirtyCardQ_CBL_mon,
+                                                  &bs->dirty_card_queue_buffer_allocator(),
+                                                  -1, // temp. never trigger
+                                                  -1, // temp. no limit
+                                                  Shared_DirtyCardQ_lock,
+                                                  true); // init_free_ids
+
+  dirty_card_queue_set().initialize(DirtyCardQ_CBL_mon,
+                                    &bs->dirty_card_queue_buffer_allocator(),
+                                    -1, // never trigger processing
+                                    -1, // no limit on length
+                                    Shared_DirtyCardQ_lock);
+
   // Create the hot card cache.
   _hot_card_cache = new G1HotCardCache(this);
 
@@ -1749,13 +1771,6 @@ jint G1CollectedHeap::initialize() {
   // Perform any initialization actions delegated to the policy.
   g1_policy()->init(this, &_collection_set);
 
-  G1BarrierSet::satb_mark_queue_set().initialize(this,
-                                                 SATB_Q_CBL_mon,
-                                                 SATB_Q_FL_lock,
-                                                 G1SATBProcessCompletedThreshold,
-                                                 G1SATBBufferEnqueueingThresholdPercent,
-                                                 Shared_SATB_Q_lock);
-
   jint ecode = initialize_concurrent_refinement();
   if (ecode != JNI_OK) {
     return ecode;
@@ -1766,20 +1781,11 @@ jint G1CollectedHeap::initialize() {
     return ecode;
   }
 
-  G1BarrierSet::dirty_card_queue_set().initialize(DirtyCardQ_CBL_mon,
+  {
-                                                  DirtyCardQ_FL_lock,
+    DirtyCardQueueSet& dcqs = G1BarrierSet::dirty_card_queue_set();
-                                                  (int)concurrent_refine()->yellow_zone(),
+    dcqs.set_process_completed_threshold((int)concurrent_refine()->yellow_zone());
-                                                  (int)concurrent_refine()->red_zone(),
+    dcqs.set_max_completed_queue((int)concurrent_refine()->red_zone());
-                                                  Shared_DirtyCardQ_lock,
+  }
-                                                  NULL,  // fl_owner
-                                                  true); // init_free_ids
-
-  dirty_card_queue_set().initialize(DirtyCardQ_CBL_mon,
-                                    DirtyCardQ_FL_lock,
-                                    -1, // never trigger processing
-                                    -1, // no limit on length
-                                    Shared_DirtyCardQ_lock,
-                                    &G1BarrierSet::dirty_card_queue_set());
 
   // Here we allocate the dummy HeapRegion that is required by the
   // G1AllocRegion class.

src/hotspot/share/gc/g1/g1ConcurrentMark.cpp

@@ -406,9 +406,6 @@ G1ConcurrentMark::G1ConcurrentMark(G1CollectedHeap* g1h,
 
   assert(CGC_lock != NULL, "CGC_lock must be initialized");
 
-  SATBMarkQueueSet& satb_qs = G1BarrierSet::satb_mark_queue_set();
-  satb_qs.set_buffer_size(G1SATBBufferSize);
-
   _root_regions.init(_g1h->survivor(), this);
 
   if (FLAG_IS_DEFAULT(ConcGCThreads) || ConcGCThreads == 0) {

src/hotspot/share/gc/g1/g1SATBMarkQueueSet.cpp

@@ -35,11 +35,13 @@
 G1SATBMarkQueueSet::G1SATBMarkQueueSet() : _g1h(NULL) {}
 
 void G1SATBMarkQueueSet::initialize(G1CollectedHeap* g1h,
-                                    Monitor* cbl_mon, Mutex* fl_lock,
+                                    Monitor* cbl_mon,
+                                    BufferNode::Allocator* allocator,
                                     int process_completed_threshold,
                                     uint buffer_enqueue_threshold_percentage,
                                     Mutex* lock) {
-  SATBMarkQueueSet::initialize(cbl_mon, fl_lock,
+  SATBMarkQueueSet::initialize(cbl_mon,
+                               allocator,
                                process_completed_threshold,
                                buffer_enqueue_threshold_percentage,
                                lock);

src/hotspot/share/gc/g1/g1SATBMarkQueueSet.hpp

@@ -37,7 +37,8 @@ public:
   G1SATBMarkQueueSet();
 
   void initialize(G1CollectedHeap* g1h,
-                  Monitor* cbl_mon, Mutex* fl_lock,
+                  Monitor* cbl_mon,
+                  BufferNode::Allocator* allocator,
                   int process_completed_threshold,
                   uint buffer_enqueue_threshold_percentage,
                   Mutex* lock);

src/hotspot/share/gc/shared/ptrQueue.cpp

@@ -26,6 +26,7 @@
 #include "gc/shared/ptrQueue.hpp"
 #include "memory/allocation.hpp"
 #include "memory/allocation.inline.hpp"
+#include "runtime/atomic.hpp"
 #include "runtime/mutex.hpp"
 #include "runtime/mutexLocker.hpp"
 #include "runtime/thread.inline.hpp"
@@ -90,25 +91,92 @@ void BufferNode::deallocate(BufferNode* node) {
   FREE_C_HEAP_ARRAY(char, node);
 }
 
+BufferNode::Allocator::Allocator(size_t buffer_size, Mutex* lock) :
+  _buffer_size(buffer_size),
+  _lock(lock),
+  _free_list(NULL),
+  _free_count(0)
+{
+  assert(lock != NULL, "precondition");
+}
+
+BufferNode::Allocator::~Allocator() {
+  while (_free_list != NULL) {
+    BufferNode* node = _free_list;
+    _free_list = node->next();
+    BufferNode::deallocate(node);
+  }
+}
+
+size_t BufferNode::Allocator::free_count() const {
+  return Atomic::load(&_free_count);
+}
+
+BufferNode* BufferNode::Allocator::allocate() {
+  BufferNode* node = NULL;
+  {
+    MutexLockerEx ml(_lock, Mutex::_no_safepoint_check_flag);
+    node = _free_list;
+    if (node != NULL) {
+      _free_list = node->next();
+      --_free_count;
+      node->set_next(NULL);
+      node->set_index(0);
+      return node;
+    }
+  }
+  return  BufferNode::allocate(_buffer_size);
+}
+
+void BufferNode::Allocator::release(BufferNode* node) {
+  MutexLockerEx ml(_lock, Mutex::_no_safepoint_check_flag);
+  node->set_next(_free_list);
+  _free_list = node;
+  ++_free_count;
+}
+
+void BufferNode::Allocator::reduce_free_list() {
+  BufferNode* head = NULL;
+  {
+    MutexLockerEx ml(_lock, Mutex::_no_safepoint_check_flag);
+    // For now, delete half.
+    size_t remove = _free_count / 2;
+    if (remove > 0) {
+      head = _free_list;
+      BufferNode* tail = head;
+      BufferNode* prev = NULL;
+      for (size_t i = 0; i < remove; ++i) {
+        assert(tail != NULL, "free list size is wrong");
+        prev = tail;
+        tail = tail->next();
+      }
+      assert(prev != NULL, "invariant");
+      assert(prev->next() == tail, "invariant");
+      prev->set_next(NULL);
+      _free_list = tail;
+      _free_count -= remove;
+    }
+  }
+  while (head != NULL) {
+    BufferNode* next = head->next();
+    BufferNode::deallocate(head);
+    head = next;
+  }
+}
+
 PtrQueueSet::PtrQueueSet(bool notify_when_complete) :
-  _buffer_size(0),
+  _allocator(NULL),
   _cbl_mon(NULL),
   _completed_buffers_head(NULL),
   _completed_buffers_tail(NULL),
   _n_completed_buffers(0),
   _process_completed_threshold(0),
   _process_completed(false),
-  _fl_lock(NULL),
-  _buf_free_list(NULL),
-  _buf_free_list_sz(0),
-  _fl_owner(NULL),
   _all_active(false),
   _notify_when_complete(notify_when_complete),
   _max_completed_queue(0),
   _completed_queue_padding(0)
-{
+{}
-  _fl_owner = this;
-}
 
 PtrQueueSet::~PtrQueueSet() {
   // There are presently only a couple (derived) instances ever
@@ -117,59 +185,24 @@ PtrQueueSet::~PtrQueueSet() {
 }
 
 void PtrQueueSet::initialize(Monitor* cbl_mon,
-                             Mutex* fl_lock,
+                             BufferNode::Allocator* allocator,
                              int process_completed_threshold,
-                             int max_completed_queue,
+                             int max_completed_queue) {
-                             PtrQueueSet *fl_owner) {
   _max_completed_queue = max_completed_queue;
   _process_completed_threshold = process_completed_threshold;
   _completed_queue_padding = 0;
-  assert(cbl_mon != NULL && fl_lock != NULL, "Init order issue?");
+  assert(cbl_mon != NULL && allocator != NULL, "Init order issue?");
   _cbl_mon = cbl_mon;
-  _fl_lock = fl_lock;
+  _allocator = allocator;
-  _fl_owner = (fl_owner != NULL) ? fl_owner : this;
 }
 
 void** PtrQueueSet::allocate_buffer() {
-  BufferNode* node = NULL;
+  BufferNode* node = _allocator->allocate();
-  {
-    MutexLockerEx x(_fl_owner->_fl_lock, Mutex::_no_safepoint_check_flag);
-    node = _fl_owner->_buf_free_list;
-    if (node != NULL) {
-      _fl_owner->_buf_free_list = node->next();
-      _fl_owner->_buf_free_list_sz--;
-    }
-  }
-  if (node == NULL) {
-    node = BufferNode::allocate(buffer_size());
-  } else {
-    // Reinitialize buffer obtained from free list.
-    node->set_index(0);
-    node->set_next(NULL);
-  }
   return BufferNode::make_buffer_from_node(node);
 }
 
 void PtrQueueSet::deallocate_buffer(BufferNode* node) {
-  MutexLockerEx x(_fl_owner->_fl_lock, Mutex::_no_safepoint_check_flag);
+  _allocator->release(node);
-  node->set_next(_fl_owner->_buf_free_list);
-  _fl_owner->_buf_free_list = node;
-  _fl_owner->_buf_free_list_sz++;
-}
-
-void PtrQueueSet::reduce_free_list() {
-  assert(_fl_owner == this, "Free list reduction is allowed only for the owner");
-  // For now we'll adopt the strategy of deleting half.
-  MutexLockerEx x(_fl_lock, Mutex::_no_safepoint_check_flag);
-  size_t n = _buf_free_list_sz / 2;
-  for (size_t i = 0; i < n; ++i) {
-    assert(_buf_free_list != NULL,
-           "_buf_free_list_sz is wrong: " SIZE_FORMAT, _buf_free_list_sz);
-    BufferNode* node = _buf_free_list;
-    _buf_free_list = node->next();
-    _buf_free_list_sz--;
-    BufferNode::deallocate(node);
-  }
 }
 
 void PtrQueue::handle_zero_index() {
@@ -270,11 +303,6 @@ void PtrQueueSet::assert_completed_buffer_list_len_correct_locked() {
             "Completed buffer length is wrong.");
 }
 
-void PtrQueueSet::set_buffer_size(size_t sz) {
-  assert(_buffer_size == 0 && sz > 0, "Should be called only once.");
-  _buffer_size = sz;
-}
-
 // Merge lists of buffers. Notify the processing threads.
 // The source queue is emptied as a result. The queues
 // must share the monitor.

src/hotspot/share/gc/shared/ptrQueue.hpp

@@ -28,6 +28,8 @@
 #include "utilities/align.hpp"
 #include "utilities/sizes.hpp"
 
+class Mutex;
+
 // There are various techniques that require threads to be able to log
 // addresses.  For example, a generational write barrier might log
 // the addresses of modified old-generation objects.  This type supports
@@ -223,18 +225,19 @@ class BufferNode {
     return offset_of(BufferNode, _buffer);
   }
 
-public:
+AIX_ONLY(public:)               // xlC 12 on AIX doesn't implement C++ DR45.
-  BufferNode* next() const     { return _next;  }
-  void set_next(BufferNode* n) { _next = n;     }
-  size_t index() const         { return _index; }
-  void set_index(size_t i)     { _index = i; }
-
   // Allocate a new BufferNode with the "buffer" having size elements.
   static BufferNode* allocate(size_t size);
 
   // Free a BufferNode.
   static void deallocate(BufferNode* node);
 
+public:
+  BufferNode* next() const     { return _next;  }
+  void set_next(BufferNode* n) { _next = n;     }
+  size_t index() const         { return _index; }
+  void set_index(size_t i)     { _index = i; }
+
   // Return the BufferNode containing the buffer, after setting its index.
   static BufferNode* make_node_from_buffer(void** buffer, size_t index) {
     BufferNode* node =
@@ -250,6 +253,24 @@ public:
     return reinterpret_cast<void**>(
       reinterpret_cast<char*>(node) + buffer_offset());
   }
+
+  // Free-list based allocator.
+  class Allocator {
+    size_t _buffer_size;
+    Mutex* _lock;
+    BufferNode* _free_list;
+    volatile size_t _free_count;
+
+  public:
+    Allocator(size_t buffer_size, Mutex* lock);
+    ~Allocator();
+
+    size_t buffer_size() const { return _buffer_size; }
+    size_t free_count() const;
+    BufferNode* allocate();
+    void release(BufferNode* node);
+    void reduce_free_list();
+  };
 };
 
 // A PtrQueueSet represents resources common to a set of pointer queues.
@@ -257,8 +278,7 @@ public:
 // set, and return completed buffers to the set.
 // All these variables are are protected by the TLOQ_CBL_mon. XXX ???
 class PtrQueueSet {
-  // The size of all buffers in the set.
+  BufferNode::Allocator* _allocator;
-  size_t _buffer_size;
 
 protected:
   Monitor* _cbl_mon;  // Protects the fields below.
@@ -268,15 +288,6 @@ protected:
   int _process_completed_threshold;
   volatile bool _process_completed;
 
-  // This (and the interpretation of the first element as a "next"
-  // pointer) are protected by the TLOQ_FL_lock.
-  Mutex* _fl_lock;
-  BufferNode* _buf_free_list;
-  size_t _buf_free_list_sz;
-  // Queue set can share a freelist. The _fl_owner variable
-  // specifies the owner. It is set to "this" by default.
-  PtrQueueSet* _fl_owner;
-
   bool _all_active;
 
   // If true, notify_all on _cbl_mon when the threshold is reached.
@@ -307,10 +318,9 @@ protected:
   // Because of init-order concerns, we can't pass these as constructor
   // arguments.
   void initialize(Monitor* cbl_mon,
-                  Mutex* fl_lock,
+                  BufferNode::Allocator* allocator,
                   int process_completed_threshold,
-                  int max_completed_queue,
+                  int max_completed_queue);
-                  PtrQueueSet *fl_owner = NULL);
 
 public:
 
@@ -336,24 +346,14 @@ public:
 
   bool is_active() { return _all_active; }
 
-  // Set the buffer size.  Should be called before any "enqueue" operation
-  // can be called.  And should only be called once.
-  void set_buffer_size(size_t sz);
-
-  // Get the buffer size.  Must have been set.
   size_t buffer_size() const {
-    assert(_buffer_size > 0, "buffer size not set");
+    return _allocator->buffer_size();
-    return _buffer_size;
   }
 
   // Get/Set the number of completed buffers that triggers log processing.
   void set_process_completed_threshold(int sz) { _process_completed_threshold = sz; }
   int process_completed_threshold() const { return _process_completed_threshold; }
 
-  // Must only be called at a safe point.  Indicates that the buffer free
-  // list size may be reduced, if that is deemed desirable.
-  void reduce_free_list();
-
   size_t completed_buffers_num() { return _n_completed_buffers; }
 
   void merge_bufferlists(PtrQueueSet* src);

src/hotspot/share/gc/shared/satbMarkQueue.cpp

@@ -111,11 +111,12 @@ SATBMarkQueueSet::SATBMarkQueueSet() :
   _buffer_enqueue_threshold(0)
 {}
 
-void SATBMarkQueueSet::initialize(Monitor* cbl_mon, Mutex* fl_lock,
+void SATBMarkQueueSet::initialize(Monitor* cbl_mon,
+                                  BufferNode::Allocator* allocator,
                                   int process_completed_threshold,
                                   uint buffer_enqueue_threshold_percentage,
                                   Mutex* lock) {
-  PtrQueueSet::initialize(cbl_mon, fl_lock, process_completed_threshold, -1);
+  PtrQueueSet::initialize(cbl_mon, allocator, process_completed_threshold, -1);
   _shared_satb_queue.set_lock(lock);
   assert(buffer_size() != 0, "buffer size not initialized");
   // Minimum threshold of 1 ensures enqueuing of completely full buffers.

src/hotspot/share/gc/shared/satbMarkQueue.hpp

@@ -108,7 +108,8 @@ protected:
     queue->apply_filter(filter);
   }
 
-  void initialize(Monitor* cbl_mon, Mutex* fl_lock,
+  void initialize(Monitor* cbl_mon,
+                  BufferNode::Allocator* allocator,
                   int process_completed_threshold,
                   uint buffer_enqueue_threshold_percentage,
                   Mutex* lock);

test/hotspot/gtest/gc/shared/test_ptrQueueBufferAllocator.cpp

@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2018, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ *
+ */
+
+#include "precompiled.hpp"
+#include "gc/shared/ptrQueue.hpp"
+#include "runtime/mutex.hpp"
+#include "unittest.hpp"
+
+// Some basic testing of BufferNode::Allocator.
+TEST_VM(PtrQueueBufferAllocatorTest, test) {
+  Mutex m(Mutex::leaf, "PtrQueueBufferAllocatorTest",
+          false, Mutex::_safepoint_check_never);
+  BufferNode::Allocator allocator(256, &m);
+
+  // Allocate some new nodes for use in testing.
+  BufferNode* nodes[10] = {};
+  const size_t node_count = ARRAY_SIZE(nodes);
+  for (size_t i = 0; i < node_count; ++i) {
+    ASSERT_EQ(0u, allocator.free_count());
+    nodes[i] = allocator.allocate();
+    ASSERT_EQ(NULL, nodes[i]->next());
+  }
+
+  // Release the nodes, adding them to the allocator's free list.
+  for (size_t i = 0; i < node_count; ++i) {
+    ASSERT_EQ(i, allocator.free_count());
+    allocator.release(nodes[i]);
+    if (i == 0) {
+      ASSERT_EQ(NULL, nodes[i]->next());
+    } else {
+      ASSERT_EQ(nodes[i - 1], nodes[i]->next());
+    }
+  }
+
+  // Allocate nodes from the free list.
+  for (size_t i = 0; i < node_count; ++i) {
+    size_t j = node_count - i;
+    ASSERT_EQ(j, allocator.free_count());
+    ASSERT_EQ(nodes[j - 1], allocator.allocate());
+  }
+  ASSERT_EQ(0u, allocator.free_count());
+
+  // Release nodes back to the free list.
+  for (size_t i = 0; i < node_count; ++i) {
+    allocator.release(nodes[i]);
+  }
+  ASSERT_EQ(node_count, allocator.free_count());
+
+  // Destroy some nodes in the free list.
+  // We don't have a way to verify destruction, but we can at
+  // leat verify we don't crash along the way.
+  allocator.reduce_free_list();
+  // destroy allocator.
+}