diff --git a/ext/msgpack/buffer.c b/ext/msgpack/buffer.c index f6a037ff..439674ce 100644 --- a/ext/msgpack/buffer.c +++ b/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; } diff --git a/spec/cruby/buffer_spec.rb b/spec/cruby/buffer_spec.rb index 4c7f554a..006c05d6 100644 --- a/spec/cruby/buffer_spec.rb +++ b/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