Fix rmem page released while a later chunk still points into it

ext/msgpack/buffer.c

@@ -127,6 +127,15 @@ void msgpack_buffer_mark(void *ptr)
 
 bool _msgpack_buffer_shift_chunk(msgpack_buffer_t* b)
 {
+    if(b->rmem_owner == &b->head->mem) {
+        /* the chunk that owns the rmem page is going away and takes the page
+         * with it. don't carve the remaining space out of a page that is
+         * already back in the pool. */
+        b->rmem_end = NULL;
+        b->rmem_last = NULL;
+        b->rmem_owner = NULL;
+    }
+
     _msgpack_buffer_chunk_destroy(b->head);
 
     if(b->head == &b->tail) {
@@ -264,6 +273,18 @@ static inline msgpack_buffer_chunk_t* _msgpack_buffer_alloc_new_chunk(msgpack_bu
     return chunk;
 }
 
+/* b->tail is copied into nc and then rebuilt, so the rmem page pointer moves
+ * to nc. the owner has to follow it: if it kept pointing at b->tail.mem, the
+ * transfer in _msgpack_buffer_chunk_malloc would degenerate into a
+ * self-assignment and the page would be released by nc while the rebuilt tail
+ * is still reading from it. */
+static inline void _msgpack_buffer_transfer_rmem_owner(msgpack_buffer_t* b, msgpack_buffer_chunk_t* nc)
+{
+    if(b->rmem_owner == &b->tail.mem) {
+        b->rmem_owner = &nc->mem;
+    }
+}
+
 static inline void _msgpack_buffer_add_new_chunk(msgpack_buffer_t* b)
 {
     if(b->head == &b->tail) {
@@ -275,6 +296,7 @@ static inline void _msgpack_buffer_add_new_chunk(msgpack_buffer_t* b)
         msgpack_buffer_chunk_t* nc = _msgpack_buffer_alloc_new_chunk(b);
 
         *nc = b->tail;
+        _msgpack_buffer_transfer_rmem_owner(b, nc);
         b->head = nc;
         nc->next = &b->tail;
 
@@ -295,6 +317,7 @@ static inline void _msgpack_buffer_add_new_chunk(msgpack_buffer_t* b)
 
         /* rebuild tail */
         *nc = b->tail;
+        _msgpack_buffer_transfer_rmem_owner(b, nc);
         before_tail->next = nc;
         nc->next = &b->tail;
     }

spec/cruby/buffer_spec.rb

@@ -620,4 +620,49 @@
 
     expect(b1.read_all).to eq(('C' * 128).b)
   end
+
+  it "keeps an rmem page alive while a later chunk still points into it" do
+    threshold = 1024
+    long = 'L' * (threshold * 2)
+
+    b1 = MessagePack::Buffer.new(nil, write_reference_threshold: threshold)
+    b1.write('a' * 100)  # first rmem page
+    b1.write(long)       # written by reference
+    b1.write('b' * 10)   # second rmem page, left mostly unused
+    b1.write(long)       # reclaims the unused part of that page
+    b1.write('c' * 50)   # carved out of the reclaimed part
+    b1.read(100 + long.bytesize + 10 + long.bytesize)
+
+    # the pool must not hand that page to anyone else while b1 still reads from it
+    others = 6.times.map do |i|
+      b = MessagePack::Buffer.new
+      b.write(('A'.ord + i).chr * 4000)
+      b
+    end
+
+    expect(b1.read_all).to eq(('c' * 50).b)
+    expect(others.size).to eq(6)
+  end
+
+  it "does not carve a new chunk out of an rmem page it already released" do
+    threshold = 1024
+    long = 'L' * (threshold * 2)
+
+    b1 = MessagePack::Buffer.new(nil, write_reference_threshold: threshold)
+    b1.write('a' * 100)
+    b1.write(long)
+    b1.write('b' * 10)
+    b1.write(long)                     # reclaims the unused part of the page
+    b1.read(100 + long.bytesize + 10)  # releases that very page
+    b1.write('d' * 50)                 # must not be carved out of it
+
+    others = 6.times.map do |i|
+      b = MessagePack::Buffer.new
+      b.write(('a'.ord + i).chr * 4000)
+      b
+    end
+
+    expect(b1.read_all).to eq((long + ('d' * 50)).b)
+    expect(others.size).to eq(6)
+  end
 end