eSpace

In a previous post, I talked a bit about event based programming for Ruby; I mentioned the EventMachine/Asymy combo as a means of doing Asynchronous database operations, hence freeing up the Ruby runtime to do other things while it is waiting on database I/O operations. Even more, the developers need not worry about using a different programming model; with the help of Ruby Fibers we will continue to program in the same old ways while Fibers will be doing all the twisted work underneath. Very promising indeed, but one big elephant in the room was the immaturity of the current solution. Asymy is still very new and it is based on the super slow pure Ruby MySQL driver, not to mention that it is fairly incomplete as well.

So, what can we do about the elephant in the room? There is an Arabic proverb that basically says "Nothing can beat iron but iron" and this is exactly what we are going to do. Enter Postgres, the database with a realistic, unfriendly elephant mascot. Go away dolphins, a real elephant is in the room now.

Surprisingly Postgres happens to have an excellent asynchronous client API; it allows you to do almost all operations in a non blocking way. More surprisingly, the Postgres driver for Ruby covers almost all those asynchronous API calls. The driver was originally written by Matz (yes, the man himself) in 1997. It was later updated by ematsu in 1999 and we have had an update fresh in March 2008 by Jdavis. If you go through the C source code, you will find many hidden gems. The methods are fairly well documented and you will discover that the driver has a blocking method that wraps the asynchronous calls inside but it does so in a Ruby threading friendly way. This way a threaded application will not block on the Postgres SQL commands. A good thing, but I am more interested in the asynchronous side of the fence.

Let's walk through the API and see how can we use it to do non blocking database access. First, you will need to install the gem ("sudo gem install pg"), then you need to require 'pg' in your code.

One problem though before we start: the current driver has this nasty little bug that prevents you from setting the connection to nonblocking. It is actually a bug in the parameter count defined in the Ruby interface; a simple switch from 0 to 1 fixes this. To save you time and sweat, I have provided a replacement gem with the modified sources (till the bug is fixed upstream). Now let's get back to the code.

Here's how to get things started:

require 'pg'
# I have configured postgres to run in *trusted* mode
# so I don't need to supply a password
conn = PGconn.new({:host=>'localhost',:user=>'postgres',:dbname=>'evented'})
conn.setnonblocking(true)

This way our connection is ready for async operations. Now we need to start sending some sql commands to our connection. To do that we normally use the PGconn#exec method, but this method will block, waiting on postgres. So instead we will use the PGconn#send_query method. This method will return immediately, not waiting for Postgres to actually process the sql command. Here's how are going to use it.

conn.send_query("select * from users where name like '%am%'")
# the method will return immediately (or raise an exception in case of an error)

But wait, where are the results? Normally we would expect the call to return with the data. Now where is my data? The results are being processed right now at the server side. We can continue to do other things till they come. But how do we know when they arrive? It turns out that this is easy as well. The PGconn instance provides a method that returns the connection's socket descriptor; PGconn#socket that is. We retrieve that socket descriptor and wrap it in a Ruby IO object by calling

io = IO.new(conn.socket)

Now have a nice IO object that we can get notified of its activity in a select call. For the uninitiated, event based programming is done by having a tight loop that runs forever. Within this loop we check if IO events happen and if so we respond to them. One efficient way of doing so is using the Ruby Kernel#select method (which is a wrapper to the UNIX select). The select method works that way: you provide it with three lists, one for sockets that you need to read from, one for sockets that you need to write to, and the third is for errors that you are interested in. The call returns an array of the sockets that can be read/write, or nil if none is ready.

We will use select as follows:

# the method that will be called if input is ready
def process_command(conn)
 # we will detail the implementation soon
end

loop do
 # we supply a list of sockets we need to read from.
 # Only our io object in this case. we nullify
 # the other lists and we set a timeout
 res = ([io],nil,nil, 0.001)
 # of course this needs to be done in a cleaner way
 process_command(conn) unless result.nil?
end

This way, whenever there is info to read from the socket, we will not get a nil (we will get an array actually) so we can call the process command. When the process command gets called it knows that there is data in the connection to be read so it calls the PGconn#consume_input method, after which it checks to see if the conn is busy or not. If it is still busy, it does nothing (it will do in a later event). On the other hand, if the connection is not busy, then we start calling the PGconn#get_result method and append what we get to the result we got so far. We keep doing that till we get a nil result which indicates the end of the command and the readiness of the connection to accept further commands. Here is how the method will look like:

def process_command(conn)
 conn.consume_input
 unless conn.is_busy
  res, data = 0, []
  while res != nil
   res = get_result
   res.each {|d| data.push d}unless res.nil?  
  end
  #we are done, we need to put this data some where 
 end
end

Several things to be noted: first, one cannot process several commands using the same connection at once. You need several connections to achieve parallel command processing. Second, the model described above works in the twisted way; to get things working the normal way you can use Ruby Fibers (or continuations, but they apparently leak memory)

I have put together a couple of Ruby classes that implement a nonblocking connection pool and a fiber pool; you can find them here. Using those you can write code that looks like this:

This way our connection is ready for async operations. Now we need to start sending some sql commands to our connection. To do that we normally use the PGconn#exec method, but this method will block, waiting on postgres. So instead we will use the PGconn#send_query method. This method will return immediately, not waiting for Postgres to actually process the sql command. Here's how are going to use it.

require 'fiber_pool'
require 'fibered_connection_pool'

options = {:host=>'localhost',:user=>'postgres',:dbname=>'evented'}

cpool = FiberedC onnectionPool.new(options, 12)
# second param is the number of connections to spawn, defaults at 8
# note that one more connection than those will be spawned. This one
# will be used for processing blocking requests.

fpool = FiberPool.new(100)
# the number of fibers to spawn, defaults at 50

100.times do
 fpool.spawn do
  cpool.exec(some_sql_command, true) #true means async
  cpool.exec(some_other_sql_command, true)
  cpool.exec(yet_another_sql_command, true)
 end
end

# our event loop
loop do
 res = select(cpool.sockets,nil,nil,0) #check for something to read
 # IO is monkey patched to be able to hold a reference to the connection
 res.first.each{ |s|s.connection.process_command } if res
end

This works as follows: once a fiber calls cpool.exec, the query is sent to the pool for processing and the fiber is halted, giving way for another one to start processing. The other one will halt as well once it hits a cpool.exec. Later, during the event loop, you will get notifications of completion of queries (in any order) and resume the fiber associated with the finished query. Note that commands issued in the same fiber will run sequentially, while those issued from different fibers will interleave. This is effectively what is achieved by threading but without its costs.

Performance:

I am sure that my code could use some tweaking, but I am getting very good results already. During benchmarking I found out that the cost on instantiating fibers could be high (the cost of pausing and resuming is high as well, but unavoidable), so I created a pool of fibers that can be reused (a very naive implementation that can make use of lots of improvement).

I tested by issuing a group of long and short queries together. You actually provide the test program with the number of long queries and the multiplier it should use for short queries, i.e. ruby test.rb 10 20 will iterate 10 times and issue a long query then within the same iteration it will issue 20 short queries. It will do this in a blocking and then nonblocking way, reporting the time taken for each to complete and the percentage of performance increase/decrease.

I tested for 10, 50 and 100 long queries with the following multipliers (1, 2, 5, 10, 50, 100). The graph shows the performance gain for each number of queries vs the multiplier. For example, 50 long queries with a multiplier of 10 (i.e. 500 short queries) achieves a 39.6% reduction in query execution time. I have repeated many of the tests several time; the repeated tests showed consistent results so I am pretty confident of the presented results.

Here is the full list:

         Queries         Mode
Ratio  Long Short Blocking Non Blocking Advantage

:1/2   10   20    0.56     0.5          10.27%
       50   100   2.55     2.26         11.19%
       100  200   5.15     4.46         13.53%

:1/5   10   50    0.55     0.4          27.04%
       50   250   2.72     1.83         32.82%
       100  500   5.45     3.63         33.39%

:1/10  10   100   0.6      0.4          33.76%
       50   500   3.01     1.82         39.67%
       100  1000  5.9      3.65         38.13%

:1/20  10   200   0.72     0.45         38.12%
       50   1000  3.43     2.1          38.73%
       100  2000  6.83     4.33         36.53%

:1/50  10   500   0.98     0.62         36.57%
       50   2500  4.78     3.23         32.36%
       100  5000  9.74     8.68         10.93%

:1/100 10   1000  1.46     0.94         35.40%
       50   5000  7.42     5.17         30.31%
       100  10000 14.27    12.68        11.15%

The area I would like to focus on for performance tuning is the size of the fiber pool. The test is a bit sensitive to it so I believe I can gain a bit more performance with insane query counts if I optimize my fiber pool a bit. Setting the initial size to high certainly helps, but eats too much memory to make it usable.

A final note: I am playing with using this along side an EventMachine based http server. It works OK but is a cpu hog (propably due to using select in next_tick calls withing EM's event loop). I would love to be able to provide EM with a list of IO objects and a call back instead of requiring me to use it to open the connection. Nevertheless, even though in many cases the nonblocking db implementation is slower than a blocking one in the http serving arena, I managed to get ~800 req/s vs ~500 req/s for a very typical use case, a request that runs a long query followed by many short ones. Impressive to say the least. I might even try to hack EM to support the feature I need and then see what performance this could yield.

Written By:

Muhammad A. Ali (oldmoe.blogspot.com)