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.

Named Parameters in Ruby

Ever forget whether the name or email parameter is first on a method like this?

signup(name,email)

Named parameters can help out so that you don't have to remember the order of parameters. As a fringe benefit the code can read a bit nicer and you can add in new optional parameters without breaking existing code.

Ruby as a language doesn't support named parameters to functions. But it does support converting named pairs into a hash if you provide an argument for it.

def signup(params)
  name=params[:name]
  email=params[:email]
  ...
end

This takes a little more work on the function declaration but it's not too bad. Now we can call the function like this:

signup(:name=>'Me', :email=>'me@net.com')

Suppose you wanted your name parameter to be optional and default to the email parameter. You can easily set default values for one or more of your expected parameters:

def signup(params)
  email=params[:email]
  name=params[:name]||email
  ...
end

With named parameters it often behooves you to do a bit more parameter checking.

def signup(params)
  email=params[:email] || raise("email parameter is required!")
  name=params[:name]||email
  ...
end

To make all parameters optional, set a default value for your parameter to {}.

def optional(params={})

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!