Memory Leak in ActiveMessaging

ActiveMessaging is great. It allows you to easily hook up to ActiveMQ to offload all your batch processing needs. Only problem, is it eats memory like crazy. Just hookup a simple queue with a publisher and consumer, write a few hundred thousand tickets, and watch the consumer eat all your available memory (will quickly eat a couple hundred megs and go on to use more than a GB).

In the gateway.rb, there is a dispatch method that routes the message to the appropriate processor:

def dispatch(message)
@@guard.synchronize {
begin
  prepare_application
  _dispatch(message)
rescue Object => exc
  ActiveMessaging.logger.error "Dispatch exception: #{exc}"
  ActiveMessaging.logger.error exc.backtrace.join("\n\t")
  raise exc
ensure
  reset_application
end
}
end

If you comment out the prepare_application and reset_application the memory consumption stops. You can chew through millions of tickets and stay at a steady usage. Only problem is that now, ActiveRecord will not keep its MySQL connection fresh, aka you will get a MySQL::Error: Mysql has gone away

These methods seem to wedge deep in rails' dispatch foo. Somewhere in there, it is likely doing validation on the connection. So, the trick will probably be to override the process!(message) method of the base processor class, and rescue MySQL::Error and call ActiveRecord::Base.verify_active_connections! and retry.

I will update this once I can validate it to see if this fixes the stale connection issue and if I run into any other issues, or if any kind commenter leaves the answer.

Ruby Batch Processing: ActiveMQ and ActiveMessinging

Lets say you have some long running background tasks that are triggered by user actions. For example, you may want to allow users to upload a list of bookmark URLs that you will create thumbnails for.

Heres the sequence of events we are looking for.


user posts list -> rails controller -> create ticket(s) -> queue -> (offline) processor(s) do work


A message queue is a good solution for this type of setup if you want to have a number of processors that you can simply start more of to scale. ActiveMQ and ActiveMessaging using STOMP make a simple ruby/rails solution.

ActiveMQ
ActiveMessaging (A13g)

Create our processor

class ThumbnailProcessor < ApplicationProcessor
 # - using ActiveMQ STOMP extension prefetchSize 
 #   to only take 1 ticket at a time
 # - set ack to client or else prefetchSize won't 
 #   do any good
 subscribes_to :thumbnail, 
   { :activemq:prefetchSize=>1, :ack=>'client'}

 def on_message(msg)
   # stub for long running code that 
   # creates thumbnail from a url
   create_thumbnail(msg)
 end
end

Create our controller

class ThumbnailController < ApplicationController
 publishes_to :thumbnail

 def create
   # create a ticket for each url in list
   params[:urls].split.each do |url| 
     publish :thumbnail, url
   end
 end
end

Configure ActiveMessaging (config/messaging.rb)

ActiveMessaging::Gateway.define do |s|
 s.destination :thumbnail, '/queue/thumbnail'
end

Now spin up as many processors as you need (you can always start more later)

./script/poller start
./script/poller start
./script/poller start

and you are ready to roll!