use a waitgroup to wait for reserve holders of MFiles before unmapping

[carlhoerberg]

WHAT is this pull request doing?

Please fill me in.

HOW can this pull request be tested?

Specs? Manual steps? Please fill me in.

[spuun]

Two things that we probably must remove

• unmap from Queue#rm_consumer
• unmap from USR2 handler

[carlhoerberg]

Two things that we probably must remove
* unmap from Queue#rm_consumer
* unmap from USR2 handler

I don't know, both call MessageStore#unmap_segments, which doesn't unmap the last segment that is still being written to.

lavinmq/src/lavinmq/amqp/queue/message_store.cr

Lines 231 to 237 in c8ca36a

	def unmap_segments(except : Enumerable(UInt32) = StaticArray(UInt32, 0).new(0u32))
	@segments.each do |seg_id, mfile|
	next if mfile == @wfile
	next if except.includes? seg_id
	mfile.unmap
	end
	end

[spuun]

I don't know, both call MessageStore#unmap_segments, which doesn't unmap the last segment that is still being written to.

Isn't it possible that a segment that isn't yet fully replicated is unmapped? If it's a large message I think the unmap may occur in the middle of the lz4 compression.

Will these unmaps be necessary if we handle unmapping properly, which this PR hopefully solves?

[carlhoerberg]

I don't know, both call MessageStore#unmap_segments, which doesn't unmap the last segment that is still being written to.

Isn't it possible that a segment that isn't yet fully replicated is unmapped? If it's a large message I think the unmap may occur in the middle of the lz4 compression.

Will these unmaps be necessary if we handle unmapping properly, which this PR hopefully solves?

True, there's a chance for that, if the unmap is triggered just when we're rolling over to a new segment.

And yes, shouldn't be needed anymore.

[carlhoerberg]

also started to tinker with a ref counter as you did, but yes, it fast gets messy, as we return a BytesMessage from MessageStore, but it points into the mmaping.. so we don't know when the client has sent them..

class MFile
  @counter = 0
  @counter_lock = Mutex.new(:unchecked)

  def borrow : self
    @counter_lock.synchronize do
      counter = @counter
      if counter.zero?
        mmap
      end
      @counter = counter + 1
    end
    self
  end

  def unborrow : Nil
    @counter_lock.synchronize do
      counter = @counter -= 1
      if counter.zero?
        unmap
      end
    end
  end
end

[carlhoerberg]

I'm starting to long for the io_uring implementation, then we could skip mmap:ings all together..

[spuun]

I'm starting to long for the io_uring implementation, then we could skip mmap:ings all together..

Yes! But I think this PR is working pretty well? I've done some runs without crashes.

[carlhoerberg]

This is still a problem: https://github.com/cloudamqp/lavinmq/blob/e13e7af6a46ef75646184e4f3128ad562d562412/src/lavinmq/amqp/queue/message_store.cr/#L261

[spuun]

This is still a problem: https://github.com/cloudamqp/lavinmq/blob/e13e7af6a46ef75646184e4f3128ad562d562412/src/lavinmq/amqp/queue/message_store.cr/#L261

Do we need this unmap? Why not let a file be mapped as long as it's opened? I.e. only unmap in #close?

[carlhoerberg]

Do we need this unmap? Why not let a file be mapped as long as it's opened? I.e. only unmap in #close?

You can't mmap unlimited number of files, eg. in linux you're limited by vm.max_map_count. If you have a queue that you only publish to no segment would never be unmapped without this

[carlhoerberg]

This is still a problem: https://github.com/cloudamqp/lavinmq/blob/e13e7af6a46ef75646184e4f3128ad562d562412/src/lavinmq/amqp/queue/message_store.cr/#L261

This is not a problem anymore, now that MFile#unmap is safe by default, MFile#unsafe_unmap is not, it doesn't check the waitgroup