memory grows in producer

[mschneider82]

running this simple producer code will show that the memory of the process will increase continuously

require 'nsq'
producer = Nsq::Producer.new(
  nsqd: "10.0.0.1:4150",
  topic: "test#ephemeral"
)
i = 0
while $i < 30000000 do
  producer.write("simple nsq message with some bytes and text to show that memory grows up #$i")
  if $i%10000==0
    puts 'RAM USAGE: ' + `pmap #{Process.pid} | tail -1`[10,40].strip
  end
  $i +=1
end

print "finished producing...  sleeping 1min"
sleep(60)
producer.terminate

consumer:

./nsq_tail --topic test#ephemeral --nsqd-tcp-address localhost:4150

[mschneider82]

Using write_to_socket directly in def pub instead of write solves this issue and has a good speed increase! The problem must be the queue

https://github.com/wistia/nsq-ruby/blob/master/lib/nsq/connection.rb#L113

[bschwartz]

Thanks for the report and for doing some sleuthing on this @mschneider82! I should have some time later this week to dig in. And if you beat me to it and figure out the leak, happy to merge a PR ;)

[mschneider82]

@bschwartz we should do a discussion here if we really need a queue between write & write_to_socket without a queue it seems to be ~10x faster, but i don't know how this will affect if a connection may gets broken

Just found this note:

  # For indicating that the connection has died.
  # We use a Queue so we don't have to poll. Used to communicate across
  # threads (from write_loop and read_loop to connect_and_monitor).

https://github.com/wistia/nsq-ruby/blob/master/lib/nsq/connection.rb#L60-L62

[mschneider82]

After some more tests i figured out that the input thread is much more faster than the output thread to the socket. Thats why messages are increasing to the internal Queue (in memory). I can produce 3.000.000 msgs in about 30secs, only 50.000 are passed through to nsqd and the rest is Queued in memory (and transfered to nsqd later).

I recommend a synchron write without a queue

[mschneider82]

In my tests SizedQueue works like a charm, the push operation gets blocked if the Queue is full, I hardcoded size of 10.000 (same as nsqd in-memory). The 3.000.000 msgs gets pushed to nsqd in ~180secs