@@ -17,4 +17,4 @@ MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION -WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. \ No newline at end of file +WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. LICENSE @@ -2,3 +2,4 @@ require 'starling/runner' StarlingServer::Runner.run + bin/starling @@ -6,7 +6,7 @@ class Starling < MemCache ## # fetch an item from a queue. - + def get(*args) loop do response = super(*args) @@ -16,13 +16,13 @@ class Starling < MemCache end ## - # insert +value+ into +queue+. + # insert +value+ into +queue+. # # +expiry+ is expressed as a UNIX timestamp - # + # # If +raw+ is true, +value+ will not be Marshalled. If +raw+ = :yaml, +value+ # will be serialized with YAML, instead. - + def set(queue, value, expiry = 0, raw = false) retries = 0 begin lib/starling.rb @@ -4,7 +4,7 @@ module StarlingServer # This is an internal class that's used by Starling::Server to handle the # MemCache protocol and act as an interface between the Server and the # QueueCollection. - + class Handler < EventMachine::Connection DATA_PACK_FMT = "Ia*".freeze @@ -49,9 +49,9 @@ STAT queue_%s_total_items %d STAT queue_%s_logsize %d STAT queue_%s_expired_items %d\n".freeze - + @@next_session_id = 1 - + ## # Creates a new handler for the MemCache protocol that communicates with a # given client. @@ -74,10 +74,10 @@ STAT queue_%s_expired_items %d\n".freeze @server.stats[:total_connections] += 1 set_comm_inactivity_timeout @opts[:timeout] @queue_collection = @opts[:queue] - + @session_id = @@next_session_id @@next_session_id += 1 - + peer = Socket.unpack_sockaddr_in(get_peername) #@logger.debug "(#{@session_id}) New session from #{peer[1]}:#{peer[0]}" end @@ -95,7 +95,7 @@ STAT queue_%s_expired_items %d\n".freeze def process(data) data = @data_buf + data if @data_buf.size > 0 - # our only non-normal state is consuming an object's data + # our only non-normal state is consuming an object's data # when @expected_length is present if @expected_length && data.size == @expected_length response = set_data(data) @@ -124,7 +124,7 @@ STAT queue_%s_expired_items %d\n".freeze logger.debug e.backtrace.join("\n") respond GET_RESPONSE_EMPTY end - + def unbind #@logger.debug "(#{@session_id}) connection ends" end @@ -176,7 +176,7 @@ STAT queue_%s_expired_items %d\n".freeze end def stats - respond STATS_RESPONSE, + respond STATS_RESPONSE, Process.pid, # pid Time.now - @server.stats(:start_time), # uptime Time.now.to_i, # time lib/starling/handler.rb @@ -27,7 +27,7 @@ module StarlingServer # Create a new PersistentQueue at +persistence_path+/+queue_name+. # If a queue log exists at that path, the Queue will be loaded from # disk before being available for use. - + def initialize(persistence_path, queue_name, debug = false) @persistence_path = persistence_path @queue_name = queue_name @@ -53,10 +53,10 @@ module StarlingServer ## # Retrieves data from the queue. - + def pop(log_trx = true) raise NoTransactionLog if log_trx && !@trx - + begin rv = super(!log_trx) rescue ThreadError @@ -69,7 +69,7 @@ module StarlingServer ## # Safely closes the transactional queue. - + def close # Ok, yeah, this is lame, and is *technically* a race condition. HOWEVER, # the QueueCollection *should* have stopped processing requests, and I don't lib/starling/persistent_queue.rb @@ -7,12 +7,12 @@ module StarlingServer ## # QueueCollection is a proxy to a collection of PersistentQueue instances. - + class QueueCollection ## # Create a new QueueCollection at +path+ - + def initialize(path) unless File.directory?(path) && File.writable?(path) raise InaccessibleQueuePath.new(path) @@ -40,7 +40,7 @@ module StarlingServer @stats[:total_items] += 1 queue.push(data) - + return true end @@ -61,7 +61,7 @@ module StarlingServer end ## - # Returns all active queues. + # Returns all active queues. def queues(key=nil) return nil if @shutdown_mutex.locked? lib/starling/queue_collection.rb @@ -11,7 +11,7 @@ module StarlingServer def self.run new end - + def initialize parse_options @@ -36,13 +36,13 @@ module StarlingServer when "log_file" then key = "logger" end options[key.to_sym] = value - + if options[:log_level].instance_of?(String) options[:log_level] = Logger.const_get(options[:log_level]) end end end - + def parse_options self.options = { :host => '127.0.0.1', :port => 22122, @@ -62,7 +62,7 @@ module StarlingServer opts.separator "" opts.separator "Configuration:" - + opts.on("-f", "--config FILENAME", "Config file (yaml) to load") do |filename| load_config_file(filename) @@ -107,7 +107,7 @@ module StarlingServer opts.on("-L", "--log [FILE]", "Path to print debugging information.") do |log_path| options[:logger] = log_path end - + opts.on("-l", "--syslog CHANNEL", "Write logs to the syslog instead of a log file.") do |channel| options[:syslog_channel] = channel end @@ -115,7 +115,7 @@ module StarlingServer opts.on("-v", "Increase logging verbosity (may be used multiple times).") do options[:log_level] -= 1 end - + opts.on("-t", "--timeout [SECONDS]", Integer, "Time in seconds before disconnecting inactive clients (0 to disable).", "(default: #{options[:timeout]})") do |timeout| @@ -229,7 +229,7 @@ module StarlingServer end end end - + def redirect_io begin; STDIN.reopen('/dev/null'); rescue Exception; end @@ -238,7 +238,7 @@ module StarlingServer STDOUT.reopen(@log_file, "a") STDOUT.sync = true rescue Exception - begin; STDOUT.reopen('/dev/null'); rescue Exception; end + begin; STDOUT.reopen('/dev/null'); rescue Exception; end end else begin; STDOUT.reopen('/dev/null'); rescue Exception; end lib/starling/runner.rb @@ -1,2 +1,3 @@ require 'test/unit' $:.unshift(File.join(File.dirname(__FILE__), "..", "lib")) + test/test_helper.rb @@ -15,3 +15,4 @@ class TestQueueCollection < Test::Unit::TestCase } end end + test/test_persistent_queue.rb @@ -1,186 +1,186 @@ -require File.join(File.dirname(__FILE__), 'test_helper.rb') - -require 'starling/server' - -require 'fileutils' -require 'rubygems' -require 'memcache' - -class StarlingServer::PersistentQueue - remove_const :SOFT_LOG_MAX_SIZE - SOFT_LOG_MAX_SIZE = 16 * 1024 # 16 KB -end - -def safely_fork(&block) - # anti-race juice: - blocking = true - Signal.trap("USR1") { blocking = false } - - pid = Process.fork(&block) - - while blocking - sleep 0.1 - end - - pid -end - - -class TestStarling < Test::Unit::TestCase - - def setup - begin - Dir::mkdir(tmp_path) - rescue Errno::EEXIST - end - - @server_pid = safely_fork do - server = StarlingServer::Base.new(:host => '127.0.0.1', - :port => 22133, - :path => tmp_path, - :logger => Logger.new(STDERR), - :log_level => Logger::FATAL) - Signal.trap("INT") { server.stop } - Process.kill("USR1", Process.ppid) - server.run - end - - @client = MemCache.new('127.0.0.1:22133') - end - - def teardown - Process.kill("INT", @server_pid) - Process.wait(@server_pid) - @client.reset - FileUtils.rm(Dir.glob(File.join(tmp_path, '*'))) - end - - def test_temporary_path_exists_and_is_writable - assert File.exist?(tmp_path) - assert File.directory?(tmp_path) - assert File.writable?(tmp_path) - end - - def test_set_and_get_one_entry - v = rand((2**32)-1) - assert_equal nil, @client.get('test_set_and_get_one_entry') - @client.set('test_set_and_get_one_entry', v) - assert_equal v, @client.get('test_set_and_get_one_entry') - end - - def test_set_with_expiry - v = rand((2**32)-1) - assert_equal nil, @client.get('test_set_with_expiry') - now = Time.now.to_i - @client.set('test_set_with_expiry', v + 2, now) - @client.set('test_set_with_expiry', v) - sleep(1.0) - assert_equal v, @client.get('test_set_with_expiry') - end - - def test_log_rotation - log_rotation_path = File.join(tmp_path, 'test_log_rotation') - - Dir.glob("#{log_rotation_path}*").each do |file| - File.unlink(file) rescue nil - end - assert_equal nil, @client.get('test_log_rotation') - - v = 'x' * 8192 - - @client.set('test_log_rotation', v) - assert_equal 8207, File.size(log_rotation_path) - @client.get('test_log_rotation') - - assert_equal nil, @client.get('test_log_rotation') - - @client.set('test_log_rotation', v) - assert_equal v, @client.get('test_log_rotation') - - assert_equal 1, File.size(log_rotation_path) - # rotated log should be erased after a successful roll. - assert_equal 1, Dir.glob("#{log_rotation_path}*").size - end - - def test_stats - stats = @client.stats - assert_kind_of Hash, stats - assert stats.has_key?('127.0.0.1:22133') - - server_stats = stats['127.0.0.1:22133'] - - basic_stats = %w( bytes pid time limit_maxbytes cmd_get version - bytes_written cmd_set get_misses total_connections - curr_connections curr_items uptime get_hits total_items - rusage_system rusage_user bytes_read ) - - basic_stats.each do |stat| - assert server_stats.has_key?(stat) - end - end - - def test_unknown_command_returns_valid_result - response = @client.add('blah', 1) - assert_match 'CLIENT_ERROR', response - end - - def test_that_disconnecting_and_reconnecting_works - v = rand(2**32-1) - @client.set('test_that_disconnecting_and_reconnecting_works', v) - @client.reset - assert_equal v, @client.get('test_that_disconnecting_and_reconnecting_works') - end - - def test_epoll - # this may take a few seconds. - # the point is to make sure that we're using epoll on Linux, so we can - # handle more than 1024 connections. - - unless IO::popen("uname").read.chomp == "Linux" - puts "(Skipping epoll test: not on Linux)" - return - end - fd_limit = IO::popen("bash -c 'ulimit -n'").read.chomp.to_i - unless fd_limit > 1024 - puts "(Skipping epoll test: 'ulimit -n' = #{fd_limit}, need > 1024)" - return - end - - v = rand(2**32 - 1) - @client.set('test_epoll', v) - - # we can't open 1024 connections to memcache from within this process, - # because we will hit ruby's 1024 fd limit ourselves! - pid1 = safely_fork do - unused_sockets = [] - 600.times do - unused_sockets << TCPSocket.new("127.0.0.1", 22133) - end - Process.kill("USR1", Process.ppid) - sleep 90 - end - pid2 = safely_fork do - unused_sockets = [] - 600.times do - unused_sockets << TCPSocket.new("127.0.0.1", 22133) - end - Process.kill("USR1", Process.ppid) - sleep 90 - end - - begin - client = MemCache.new('127.0.0.1:22133') - assert_equal v, client.get('test_epoll') - ensure - Process.kill("TERM", pid1) - Process.kill("TERM", pid2) - end - end - - - private - - def tmp_path - File.join(File.dirname(__FILE__), "..", "tmp") - end -end +require File.join(File.dirname(__FILE__), 'test_helper.rb') + +require 'starling/server' + +require 'fileutils' +require 'rubygems' +require 'memcache' + +class StarlingServer::PersistentQueue + remove_const :SOFT_LOG_MAX_SIZE + SOFT_LOG_MAX_SIZE = 16 * 1024 # 16 KB +end + +def safely_fork(&block) + # anti-race juice: + blocking = true + Signal.trap("USR1") { blocking = false } + + pid = Process.fork(&block) + + while blocking + sleep 0.1 + end + + pid +end + + +class TestStarling < Test::Unit::TestCase + + def setup + begin + Dir::mkdir(tmp_path) + rescue Errno::EEXIST + end + + @server_pid = safely_fork do + server = StarlingServer::Base.new(:host => '127.0.0.1', + :port => 22133, + :path => tmp_path, + :logger => Logger.new(STDERR), + :log_level => Logger::FATAL) + Signal.trap("INT") { server.stop } + Process.kill("USR1", Process.ppid) + server.run + end + + @client = MemCache.new('127.0.0.1:22133') + end + + def teardown + Process.kill("INT", @server_pid) + Process.wait(@server_pid) + @client.reset + FileUtils.rm(Dir.glob(File.join(tmp_path, '*'))) + end + + def test_temporary_path_exists_and_is_writable + assert File.exist?(tmp_path) + assert File.directory?(tmp_path) + assert File.writable?(tmp_path) + end + + def test_set_and_get_one_entry + v = rand((2**32)-1) + assert_equal nil, @client.get('test_set_and_get_one_entry') + @client.set('test_set_and_get_one_entry', v) + assert_equal v, @client.get('test_set_and_get_one_entry') + end + + def test_set_with_expiry + v = rand((2**32)-1) + assert_equal nil, @client.get('test_set_with_expiry') + now = Time.now.to_i + @client.set('test_set_with_expiry', v + 2, now) + @client.set('test_set_with_expiry', v) + sleep(1.0) + assert_equal v, @client.get('test_set_with_expiry') + end + + def test_log_rotation + log_rotation_path = File.join(tmp_path, 'test_log_rotation') + + Dir.glob("#{log_rotation_path}*").each do |file| + File.unlink(file) rescue nil + end + assert_equal nil, @client.get('test_log_rotation') + + v = 'x' * 8192 + + @client.set('test_log_rotation', v) + assert_equal 8207, File.size(log_rotation_path) + @client.get('test_log_rotation') + + assert_equal nil, @client.get('test_log_rotation') + + @client.set('test_log_rotation', v) + assert_equal v, @client.get('test_log_rotation') + + assert_equal 1, File.size(log_rotation_path) + # rotated log should be erased after a successful roll. + assert_equal 1, Dir.glob("#{log_rotation_path}*").size + end + + def test_stats + stats = @client.stats + assert_kind_of Hash, stats + assert stats.has_key?('127.0.0.1:22133') + + server_stats = stats['127.0.0.1:22133'] + + basic_stats = %w( bytes pid time limit_maxbytes cmd_get version + bytes_written cmd_set get_misses total_connections + curr_connections curr_items uptime get_hits total_items + rusage_system rusage_user bytes_read ) + + basic_stats.each do |stat| + assert server_stats.has_key?(stat) + end + end + + def test_unknown_command_returns_valid_result + response = @client.add('blah', 1) + assert_match 'CLIENT_ERROR', response + end + + def test_that_disconnecting_and_reconnecting_works + v = rand(2**32-1) + @client.set('test_that_disconnecting_and_reconnecting_works', v) + @client.reset + assert_equal v, @client.get('test_that_disconnecting_and_reconnecting_works') + end + + def test_epoll + # this may take a few seconds. + # the point is to make sure that we're using epoll on Linux, so we can + # handle more than 1024 connections. + + unless IO::popen("uname").read.chomp == "Linux" + puts "(Skipping epoll test: not on Linux)" + return + end + fd_limit = IO::popen("bash -c 'ulimit -n'").read.chomp.to_i + unless fd_limit > 1024 + puts "(Skipping epoll test: 'ulimit -n' = #{fd_limit}, need > 1024)" + return + end + + v = rand(2**32 - 1) + @client.set('test_epoll', v) + + # we can't open 1024 connections to memcache from within this process, + # because we will hit ruby's 1024 fd limit ourselves! + pid1 = safely_fork do + unused_sockets = [] + 600.times do + unused_sockets << TCPSocket.new("127.0.0.1", 22133) + end + Process.kill("USR1", Process.ppid) + sleep 90 + end + pid2 = safely_fork do + unused_sockets = [] + 600.times do + unused_sockets << TCPSocket.new("127.0.0.1", 22133) + end + Process.kill("USR1", Process.ppid) + sleep 90 + end + + begin + client = MemCache.new('127.0.0.1:22133') + assert_equal v, client.get('test_epoll') + ensure + Process.kill("TERM", pid1) + Process.kill("TERM", pid2) + end + end + + + private + + def tmp_path + File.join(File.dirname(__FILE__), "..", "tmp") + end +end test/test_starling.rb 4c606a4ece4f5a17b49fb2d40d627f594d1d50b8 Glenn Rempe glenn@rempe.us http://github.com/defunkt/starling/commit/7a0b965e3953539741a94550ca4af59ae7beb7b9 7a0b965e3953539741a94550ca4af59ae7beb7b9 2008-04-28T17:26:27-07:00 2008-04-28T17:26:27-07:00 Whitespace cleanup. Removed trailing whitespace. a296146c069b0ece40ca3c44fdc461814ce5ff9a Glenn Rempe glenn@rempe.us