Speed up access to nested buffers

get_last_nested() used to walk the entire list, requiring linear time.
Because get_last_nested() is also invoked a linear amount of times, this
leads to an overall quadratic running time in order to access all nested
buffers. This effect is called 'accidentally quadratic', and is a common
issue.

The 'Can quickly handle deeply nested buffer' test makes sure that this
issue does not regress in the future, and also allows me to make some
guesstimates about the performance gain: For a nesting depth of 30000,
the test is sped up from 4 seconds to less than 0.01 seconds.

Here are some more practical numbers: Iterating over the entire planet
(which has very large blocks and thus deeply nested buffers), I observe
a speed up from 163.1 s to 160.9 s according to hyperfine (10 runs each).

include/osmium/memory/buffer.hpp

@@ -103,6 +103,29 @@ namespace osmium {
 
         private:
 
+            /**
+             * The fields m_prev_buffer_or_end and m_next_buffer effectively are
+             * roughly a intrusive doubly-linked list, except that the head is
+             * never pointed-to. m_prev_buffer_or_end is necessary to quickly get
+             * access to the last buffer in a long chain, and prevent an
+             * accidentally-quadratic time. The head cannot be pointed-to, as it
+             * can be move-assigned to a different location in memory.
+             *
+             * Note that the caller never gains access to any of the nested buffers
+             * directly. In particular, get_last_nested() pops the last nested
+             * Buffer, meaning it is no longer present in this chain. This reduces
+             * the possible scenarios that need to be handled.
+             *
+             * - If this Buffer does not participate in any nesting,
+             *   then m_prev_buffer_or_end is nullptr.
+             * - If this Buffer is the head (or start) of the nested Buffer chain,
+             *   then m_prev_buffer_or_end points to the last nested Buffer.
+             * - If this Buffer is part of a nesting chain, but not the head of
+             *   the chain, then m_prev_buffer_or_end points to the previous
+             *   *nested* Buffer in the chain (or nullptr, if the previous Buffer
+             *   would be the head).
+             */
+            Buffer* m_prev_buffer_or_end = nullptr;
             std::unique_ptr<Buffer> m_next_buffer;
             std::unique_ptr<unsigned char[]> m_memory{};
             unsigned char* m_data = nullptr;
@@ -111,6 +134,7 @@ namespace osmium {
             std::size_t m_committed = 0;
 #ifndef NDEBUG
             uint8_t m_builder_count = 0;
+            bool m_this_is_nesting_list_head = true;
 #endif
             auto_grow m_auto_grow{auto_grow::no};
 
@@ -133,15 +157,28 @@ namespace osmium {
                 }
 
                 std::unique_ptr<Buffer> old{new Buffer{std::move(m_memory), m_capacity, m_committed}};
+#ifndef NDEBUG
+                assert(m_this_is_nesting_list_head);
+                old->m_this_is_nesting_list_head = false;
+#endif
                 m_memory = std::unique_ptr<unsigned char[]>{new unsigned char[m_capacity]};
                 m_data = m_memory.get();
 
                 m_written -= m_committed;
                 std::copy_n(old->data() + m_committed, m_written, m_data);
                 m_committed = 0;
 
-                old->m_next_buffer = std::move(m_next_buffer);
+                if (m_next_buffer) {
+                    // Link "old" and "m_next_buffer":
+                    m_next_buffer->m_prev_buffer_or_end = old.get();
+                    old->m_next_buffer = std::move(m_next_buffer);
+                } else {
+                    m_prev_buffer_or_end = old.get();
+                }
+
+                // Link "this" and "old":
                 m_next_buffer = std::move(old);
+                // Intentionally don't write m_next_buffer->m_prev_buffer_or_end here!
             }
 
         public:
@@ -259,6 +296,7 @@ namespace osmium {
 
             // buffers can be moved
             Buffer(Buffer&& other) noexcept :
+                m_prev_buffer_or_end(std::move(other.m_prev_buffer_or_end)),
                 m_next_buffer(std::move(other.m_next_buffer)),
                 m_memory(std::move(other.m_memory)),
                 m_data(other.m_data),
@@ -279,6 +317,7 @@ namespace osmium {
             }
 
             Buffer& operator=(Buffer&& other) noexcept {
+                m_prev_buffer_or_end = std::move(other.m_prev_buffer_or_end);
                 m_next_buffer = std::move(other.m_next_buffer);
                 m_memory = std::move(other.m_memory);
                 m_data = other.m_data;
@@ -400,6 +439,9 @@ namespace osmium {
              * @returns Are there nested buffers or not?
              */
             bool has_nested_buffers() const noexcept {
+#ifndef NDEBUG
+                assert(m_this_is_nesting_list_head);
+#endif
                 return m_next_buffer != nullptr;
             }
 
@@ -411,11 +453,25 @@ namespace osmium {
              */
             std::unique_ptr<Buffer> get_last_nested() {
                 assert(has_nested_buffers());
-                Buffer* buffer = this;
-                while (buffer->m_next_buffer->has_nested_buffers()) {
-                    buffer = buffer->m_next_buffer.get();
+#ifndef NDEBUG
+                assert(m_this_is_nesting_list_head);
+#endif
+                Buffer* last_buffer = m_prev_buffer_or_end;
+                m_prev_buffer_or_end = last_buffer->m_prev_buffer_or_end;
+                std::unique_ptr<Buffer> last_buffer_owner;
+                if (Buffer* second_last_buffer = m_prev_buffer_or_end) {
+                    // Unlink second_last_buffer and last_buffer:
+                    last_buffer_owner = std::move(second_last_buffer->m_next_buffer);
+                    last_buffer->m_prev_buffer_or_end = nullptr;
+                } else {
+                    // Unlink "this" and last_buffer:
+                    last_buffer_owner = std::move(m_next_buffer);
+                    m_prev_buffer_or_end = nullptr;
                 }
-                return std::move(buffer->m_next_buffer);
+#ifndef NDEBUG
+                last_buffer->m_this_is_nesting_list_head = true;
+#endif
+                return last_buffer_owner;
             }
 
             /**

test/t/memory/test_buffer_nested.cpp

@@ -59,7 +59,18 @@ TEST_CASE("Can populate and iterate auto_grow::yes") {
     buffer.select<osmium::Node>();
 }
 
-TEST_CASE("Can populate but NOT iterate nested auto_grow::internal") {
+static void require_movable(osmium::memory::Buffer& buffer) {
+    osmium::memory::Buffer move_construct_buffer{std::move(buffer)};
+    REQUIRE_FALSE(buffer);
+    REQUIRE(move_construct_buffer);
+    osmium::memory::Buffer move_assign_buffer;
+    move_assign_buffer = std::move(move_construct_buffer);
+    REQUIRE_FALSE(buffer);
+    REQUIRE_FALSE(move_construct_buffer);
+    REQUIRE(move_assign_buffer);
+}
+
+TEST_CASE("Can populate but NOT iterate nested auto_grow::internal, and de-nest it") {
     osmium::memory::Buffer buffer = make_populated_buffer(10, osmium::memory::Buffer::auto_grow::internal);
     REQUIRE(buffer);
     REQUIRE(buffer.is_aligned());
@@ -81,6 +92,35 @@ TEST_CASE("Can populate but NOT iterate nested auto_grow::internal") {
     // REQUIRE_ASSERT(buffer.get_iterator(0));
     // REQUIRE_ASSERT(buffer.get_iterator<osmium::Node>(0));
     // REQUIRE_ASSERT(buffer.select<osmium::Node>());
+
+    SECTION("Can move nested buffer") {
+        require_movable(buffer);
+    }
+    SECTION("Can de-nest buffer") {
+        REQUIRE(buffer.has_nested_buffers());
+        std::unique_ptr<osmium::memory::Buffer> first_buffer = buffer.get_last_nested();
+        REQUIRE(!first_buffer->has_nested_buffers());
+        REQUIRE(first_buffer->capacity() == 512);
+        REQUIRE(first_buffer->written() > 0);
+        REQUIRE(first_buffer->written() < 512);
+        REQUIRE(first_buffer->committed() == first_buffer->written());
+        first_buffer->begin();
+        require_movable(*first_buffer);
+        REQUIRE(buffer.has_nested_buffers());
+
+        SECTION("Can move partially de-nested buffer") {
+            require_movable(buffer);
+        }
+        SECTION("Can further de-nest buffer") {
+            std::unique_ptr<osmium::memory::Buffer> second_buffer = buffer.get_last_nested();
+            second_buffer->begin();
+            REQUIRE(buffer.has_nested_buffers());
+            std::unique_ptr<osmium::memory::Buffer> third_buffer = buffer.get_last_nested();
+            third_buffer->begin();
+            REQUIRE(!buffer.has_nested_buffers());
+            buffer.begin();
+        }
+    }
 }
 
 TEST_CASE("Can populate and iterate unnested auto_grow::internal") {
@@ -105,6 +145,9 @@ TEST_CASE("Can populate and iterate unnested auto_grow::internal") {
     buffer.get_iterator(0);
     buffer.get_iterator<osmium::Node>(0);
     buffer.select<osmium::Node>();
+
+    // REQUIRE that the buffer can be moved:
+    require_movable(buffer);
 }
 
 TEST_CASE("Can iterate empty auto_grow::internal") {
@@ -128,4 +171,26 @@ TEST_CASE("Can iterate empty auto_grow::internal") {
     buffer.get_iterator(0);
     buffer.get_iterator<osmium::Node>(0);
     buffer.select<osmium::Node>();
+
+    // REQUIRE that the buffer can be moved:
+    require_movable(buffer);
+}
+
+TEST_CASE("Can quickly handle deeply nested buffer") {
+    osmium::memory::Buffer buffer = make_populated_buffer(100000, osmium::memory::Buffer::auto_grow::internal);
+    REQUIRE(buffer);
+    REQUIRE(buffer.is_aligned());
+    REQUIRE(buffer.capacity() == 512);
+    REQUIRE(buffer.written() > 0);
+    REQUIRE(buffer.written() < 512);
+    REQUIRE(buffer.committed() == buffer.written());
+    REQUIRE(buffer.has_nested_buffers());
+
+    std::vector<std::unique_ptr<osmium::memory::Buffer>> sub_buffers;
+    while (buffer.has_nested_buffers()) {
+        sub_buffers.push_back(buffer.get_last_nested());
+    }
+    REQUIRE(sub_buffers.size() > 30000);
+    // If get_last_nested() runs in constant time, then this test case is lightning fast (< 0.01s).
+    // If get_last_nested() needs to walk the entire list, then the run time is really slow (~ 4s).
 }