Bug fix: correct frame-pointer wraparound when processing batch

basic_log::output_worker calculates frame positions by taking the
"front" address (current read pointer) from mpsc_ring_buffer and then
advancing that until it reaches the end of the batch. However, if the
batch size is large enough to go past the end of the ring buffer then no
wrap around is applied to the frame pointer.

This isn't strictly a buffer overrun because, due to the nature of the
ring buffer implementation, the same physical memory is mapped in a
second time at the end and we are in fact meant to go past the end in
situations where objects overlap the end. However, we can run into
trouble when an object was constructed in the first virtual memory-map
and then accessed through the second virtual-memory map. For plain old
data it works fine, but if the object has code that depends on the
object's this-pointer staying consistent (a reasonable expectation) then
formatting or destroying the object can lead to undefined behavior.

This surfaced as github issue #36 in which the libstdc++ implementation
of std::string had optimizations that would make use of the address of a
class member to determine which deallocation strategy must be used.

The fix is fairly simple: instead of using the frame pointer as loop
invariant we use the ring buffer's 64-bit position to determine when the
end is reached, and whenever we need to use the frame pointer we make
sure that it is consistently wrapped to always start in the first
virtual-memory block.

reckless/include/reckless/detail/mpsc_ring_buffer.hpp

@@ -98,9 +98,14 @@ class mpsc_ring_buffer {
         }
     }
 
-    void* front() noexcept
+    std::uint64_t read_position() noexcept
     {
-        return pbuffer_start_ + (next_read_position_ & (capacity_ - 1));
+        return next_read_position_;
+    }
+
+    void* address(std::uint64_t position) noexcept
+    {
+        return pbuffer_start_ + (position & (capacity_ - 1));
     }
 
     std::size_t size() noexcept

reckless/src/basic_log.cpp

@@ -388,23 +388,24 @@ void basic_log::output_worker()
             std::max(input_buffer_high_watermark_, batch_size));
         RECKLESS_TRACE(process_batch_start_event, batch_size);
 
-        auto pbatch_start = static_cast<char*>(input_buffer_.front());
-        auto pbatch_end = pbatch_start + batch_size;
-        auto pframe = pbatch_start;
+        auto batch_start = input_buffer_.read_position();
+        auto batch_end = batch_start + batch_size;
+        auto read_position = batch_start;
 
         bool panic_flush = false;
         do
         {
-            while(pframe != pbatch_end &&
+            while(read_position != batch_end &&
                  likely(status < frame_status::shutdown_marker))
             {
+                void* pframe = input_buffer_.address(read_position);
                 status = acquire_frame(pframe);
 
                 std::size_t frame_size;
                 if(likely(status == frame_status::initialized))
                     frame_size = process_frame(pframe);
                 else if(status == frame_status::failed_error_check)
-                    frame_size = char_cast<frame_header*>(pframe)->frame_size;
+                    frame_size = static_cast<frame_header*>(pframe)->frame_size;
                 else if(status == frame_status::failed_initialization)
                     frame_size = skip_frame(pframe);
                 else if(status == frame_status::shutdown_marker)
@@ -417,9 +418,9 @@ void basic_log::output_worker()
                 }
 
                 clear_frame(pframe, frame_size);
-                pframe += frame_size;
+                read_position += frame_size;
             }
-            assert(pframe == pbatch_end);
+            assert(read_position == batch_end);
 
             // Return memory to the input buffer to be used by other threads,
             // but only if we are not in a panic-flush state.
@@ -430,15 +431,15 @@ void basic_log::output_worker()
             } else {
                 // As a consequence of not returning memory to the input buffer,
                 // on the next batch iteration input_buffer_.front() is going
-                // to return exactly the same frame address that we already
+                // to return exactly the same position that we already
                 // processed, meaning we will hang waiting for it to become
                 // initialized. So instead of continuing normally we just update
                 // the batch end to reflect the current size of the buffer
                 // (which is going to end up equal to the full capacity of the
                 // buffer), let pframe remain at its current position, and loop
                 // around until we reach the panic_shutdown_marker frame.
                 batch_size = input_buffer_.size();
-                pbatch_end = pbatch_start + batch_size;
+                batch_end = batch_start + batch_size;
             }
         } while(unlikely(panic_flush));
         RECKLESS_TRACE(process_batch_finish_event);

tests/consistent_address.cpp

@@ -0,0 +1,67 @@
+/* This file is part of reckless logging
+ * Copyright 2015-2020 Mattias Flodin <[email protected]>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+// There was a bug in the pointer arithmetic in basic_log that would sometimes
+// cause a different virtual address (but the same physical address) than was
+// used during construction to be used for destroying an object submitted to
+// the log. This test case reproduces that bug.
+
+#include <reckless/policy_log.hpp>
+#include <reckless/file_writer.hpp>
+
+#include <cstdio>
+#include <cstdlib>
+
+struct holder {
+    void* pmy_address;
+    holder()
+    {
+        pmy_address = this;
+    }
+    holder(const holder&)
+    {
+        pmy_address = this;
+    }
+    ~holder()
+    {
+        if (this != pmy_address) {
+            std::fprintf(stderr, "inconsistent object address; expected %p "
+                "but got %p\n", pmy_address, this);
+            std::fflush(stderr);
+            std::abort();
+        }
+    }
+};
+
+char const* format(reckless::output_buffer* p, char const* fmt, holder const&)
+{
+    return fmt+1;
+}
+
+int main()
+{
+    reckless::file_writer writer("log.txt");
+    reckless::policy_log<> log(&writer);
+    for(int i=0; i!=500000; i++) {
+        log.write("%s", holder());
+    }
+}