lib/rb/benchmark/benchmark.rb - thrift - Git at Google

 #
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements. See the NOTICE file
 # distributed with this work for additional information
 # regarding copyright ownership. The ASF licenses this file
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License. You may obtain a copy of the License at
 #
 #   http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 # KIND, either express or implied. See the License for the
 # specific language governing permissions and limitations
 # under the License.
 #

 require 'rubygems'
 $:.unshift File.dirname(__FILE__) + '/../lib'
 require 'thrift'
 require 'stringio'

 HOST = '127.0.0.1'
 PORT = 42587

 ###############
 ## Server
 ###############

 class Server
   attr_accessor :serverclass
   attr_accessor :interpreter
   attr_accessor :host
   attr_accessor :port

   def initialize(opts)
     @serverclass = opts.fetch(:class, Thrift::NonblockingServer)
     @interpreter = opts.fetch(:interpreter, "ruby")
     @host = opts.fetch(:host, ::HOST)
     @port = opts.fetch(:port, ::PORT)
   end

   def start
     return if @serverclass == Object
     args = (File.basename(@interpreter) == "jruby" ? "-J-server" : "")
     @pipe = IO.popen("#{@interpreter} #{args} #{File.dirname(__FILE__)}/server.rb #{@host} #{@port} #{@serverclass.name}", "r+")
     Marshal.load(@pipe) # wait until the server has started
     sleep 0.4 # give the server time to actually start spawning sockets
   end

   def shutdown
     return unless @pipe
     Marshal.dump(:shutdown, @pipe)
     begin
       @pipe.read(10) # block until the server shuts down
     rescue EOFError
     end
     @pipe.close
     @pipe = nil
   end
 end

 class BenchmarkManager
   def initialize(opts, server)
     @socket = opts.fetch(:socket) do
       @host = opts.fetch(:host, 'localhost')
       @port = opts.fetch(:port)
       nil
     end
     @num_processes = opts.fetch(:num_processes, 40)
     @clients_per_process = opts.fetch(:clients_per_process, 10)
     @calls_per_client = opts.fetch(:calls_per_client, 50)
     @interpreter = opts.fetch(:interpreter, "ruby")
     @server = server
     @log_exceptions = opts.fetch(:log_exceptions, false)
   end

   def run
     @pool = []
     @benchmark_start = Time.now
     puts "Spawning benchmark processes..."
     @num_processes.times do
       spawn
       sleep 0.02 # space out spawns
     end
     collect_output
     @benchmark_end = Time.now # we know the procs are done here
     translate_output
     analyze_output
     report_output
   end

   def spawn
     pipe = IO.popen("#{@interpreter} #{File.dirname(__FILE__)}/client.rb #{"-log-exceptions" if @log_exceptions} #{@host} #{@port} #{@clients_per_process} #{@calls_per_client}")
     @pool << pipe
   end

   def socket_class
     if @socket
       Thrift::UNIXSocket
     else
       Thrift::Socket
     end
   end

   def collect_output
     puts "Collecting output..."
     # read from @pool until all sockets are closed
     @buffers = Hash.new { |h,k| h[k] = '' }
     until @pool.empty?
       rd, = select(@pool)
       next if rd.nil?
       rd.each do |fd|
         begin
           @buffers[fd] << fd.readpartial(4096)
         rescue EOFError
           @pool.delete fd
         end
       end
     end
   end

   def translate_output
     puts "Translating output..."
     @output = []
     @buffers.each do |fd, buffer|
       strio = StringIO.new(buffer)
       logs = []
       begin
         loop do
           logs << Marshal.load(strio)
         end
       rescue EOFError
         @output << logs
       end
     end
   end

   def analyze_output
     puts "Analyzing output..."
     call_times = []
     client_times = []
     connection_failures = []
     connection_errors = []
     shortest_call = 0
     shortest_client = 0
     longest_call = 0
     longest_client = 0
     @output.each do |logs|
       cur_call, cur_client = nil
       logs.each do |tok, time|
         case tok
         when :start
           cur_client = time
         when :call_start
           cur_call = time
         when :call_end
           delta = time - cur_call
           call_times << delta
           longest_call = delta unless longest_call > delta
           shortest_call = delta if shortest_call == 0 or delta < shortest_call
           cur_call = nil
         when :end
           delta = time - cur_client
           client_times << delta
           longest_client = delta unless longest_client > delta
           shortest_client = delta if shortest_client == 0 or delta < shortest_client
           cur_client = nil
         when :connection_failure
           connection_failures << time
         when :connection_error
           connection_errors << time
         end
       end
     end
     @report = {}
     @report[:total_calls] = call_times.inject(0.0) { |a,t| a += t }
     @report[:avg_calls] = @report[:total_calls] / call_times.size
     @report[:total_clients] = client_times.inject(0.0) { |a,t| a += t }
     @report[:avg_clients] = @report[:total_clients] / client_times.size
     @report[:connection_failures] = connection_failures.size
     @report[:connection_errors] = connection_errors.size
     @report[:shortest_call] = shortest_call
     @report[:shortest_client] = shortest_client
     @report[:longest_call] = longest_call
     @report[:longest_client] = longest_client
     @report[:total_benchmark_time] = @benchmark_end - @benchmark_start
     @report[:fastthread] = $".include?('fastthread.bundle')
   end

   def report_output
     fmt = "%.4f seconds"
     puts
     tabulate "%d",
              [["Server class", "%s"], @server.serverclass == Object ? "" : @server.serverclass],
              [["Server interpreter", "%s"], @server.interpreter],
              [["Client interpreter", "%s"], @interpreter],
              [["Socket class", "%s"], socket_class],
              ["Number of processes", @num_processes],
              ["Clients per process", @clients_per_process],
              ["Calls per client", @calls_per_client],
              [["Using fastthread", "%s"], @report[:fastthread] ? "yes" : "no"]
     puts
     failures = (@report[:connection_failures] > 0)
     tabulate fmt,
              [["Connection failures", "%d", [:red, :bold]], @report[:connection_failures]],
              [["Connection errors", "%d", [:red, :bold]], @report[:connection_errors]],
              ["Average time per call", @report[:avg_calls]],
              ["Average time per client (%d calls)" % @calls_per_client, @report[:avg_clients]],
              ["Total time for all calls", @report[:total_calls]],
              ["Real time for benchmarking", @report[:total_benchmark_time]],
              ["Shortest call time", @report[:shortest_call]],
              ["Longest call time", @report[:longest_call]],
              ["Shortest client time (%d calls)" % @calls_per_client, @report[:shortest_client]],
              ["Longest client time (%d calls)" % @calls_per_client, @report[:longest_client]]
   end

   ANSI = {
     :reset => 0,
     :bold => 1,
     :black => 30,
     :red => 31,
     :green => 32,
     :yellow => 33,
     :blue => 34,
     :magenta => 35,
     :cyan => 36,
     :white => 37
   }

   def tabulate(fmt, *labels_and_values)
     labels = labels_and_values.map { |l| Array === l ? l.first : l }
     label_width = labels.inject(0) { |w,l| l.size > w ? l.size : w }
     labels_and_values.each do |(l,v)|
       f = fmt
       l, f, c = l if Array === l
       fmtstr = "%-#{label_width+1}s #{f}"
       if STDOUT.tty? and c and v.to_i > 0
         fmtstr = "\e[#{[*c].map { |x| ANSI[x] } * ";"}m" + fmtstr + "\e[#{ANSI[:reset]}m"
       end
       puts fmtstr % [l+":", v]
     end
   end
 end

 def resolve_const(const)
   const and const.split('::').inject(Object) { |k,c| k.const_get(c) }
 end

 puts "Starting server..."
 args = {}
 args[:interpreter] = ENV['THRIFT_SERVER_INTERPRETER'] || ENV['THRIFT_INTERPRETER'] || "ruby"
 args[:class] = resolve_const(ENV['THRIFT_SERVER']) || Thrift::NonblockingServer
 args[:host] = ENV['THRIFT_HOST'] || HOST
 args[:port] = (ENV['THRIFT_PORT'] || PORT).to_i
 server = Server.new(args)
 server.start

 args = {}
 args[:host] = ENV['THRIFT_HOST'] || HOST
 args[:port] = (ENV['THRIFT_PORT'] || PORT).to_i
 args[:num_processes] = (ENV['THRIFT_NUM_PROCESSES'] || 40).to_i
 args[:clients_per_process] = (ENV['THRIFT_NUM_CLIENTS'] || 5).to_i
 args[:calls_per_client] = (ENV['THRIFT_NUM_CALLS'] || 50).to_i
 args[:interpreter] = ENV['THRIFT_CLIENT_INTERPRETER'] || ENV['THRIFT_INTERPRETER'] || "ruby"
 args[:log_exceptions] = !!ENV['THRIFT_LOG_EXCEPTIONS']
 BenchmarkManager.new(args, server).run

 server.shutdown
	#
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.
	#

	require 'rubygems'
	$:.unshift File.dirname(__FILE__) + '/../lib'
	require 'thrift'
	require 'stringio'

	HOST = '127.0.0.1'
	PORT = 42587

	###############
	## Server
	###############

	class Server
	attr_accessor :serverclass
	attr_accessor :interpreter
	attr_accessor :host
	attr_accessor :port

	def initialize(opts)
	@serverclass = opts.fetch(:class, Thrift::NonblockingServer)
	@interpreter = opts.fetch(:interpreter, "ruby")
	@host = opts.fetch(:host, ::HOST)
	@port = opts.fetch(:port, ::PORT)
	end

	def start
	return if @serverclass == Object
	args = (File.basename(@interpreter) == "jruby" ? "-J-server" : "")
	@pipe = IO.popen("#{@interpreter} #{args} #{File.dirname(__FILE__)}/server.rb #{@host} #{@port} #{@serverclass.name}", "r+")
	Marshal.load(@pipe) # wait until the server has started
	sleep 0.4 # give the server time to actually start spawning sockets
	end

	def shutdown
	return unless @pipe
	Marshal.dump(:shutdown, @pipe)
	begin
	@pipe.read(10) # block until the server shuts down
	rescue EOFError
	end
	@pipe.close
	@pipe = nil
	end
	end

	class BenchmarkManager
	def initialize(opts, server)
	@socket = opts.fetch(:socket) do
	@host = opts.fetch(:host, 'localhost')
	@port = opts.fetch(:port)
	nil
	end
	@num_processes = opts.fetch(:num_processes, 40)
	@clients_per_process = opts.fetch(:clients_per_process, 10)
	@calls_per_client = opts.fetch(:calls_per_client, 50)
	@interpreter = opts.fetch(:interpreter, "ruby")
	@server = server
	@log_exceptions = opts.fetch(:log_exceptions, false)
	end

	def run
	@pool = []
	@benchmark_start = Time.now
	puts "Spawning benchmark processes..."
	@num_processes.times do
	spawn
	sleep 0.02 # space out spawns
	end
	collect_output
	@benchmark_end = Time.now # we know the procs are done here
	translate_output
	analyze_output
	report_output
	end

	def spawn
	pipe = IO.popen("#{@interpreter} #{File.dirname(__FILE__)}/client.rb #{"-log-exceptions" if @log_exceptions} #{@host} #{@port} #{@clients_per_process} #{@calls_per_client}")
	@pool << pipe
	end

	def socket_class
	if @socket
	Thrift::UNIXSocket
	else
	Thrift::Socket
	end
	end

	def collect_output
	puts "Collecting output..."
	# read from @pool until all sockets are closed
	@buffers = Hash.new { \|h,k\| h[k] = '' }
	until @pool.empty?
	rd, = select(@pool)
	next if rd.nil?
	rd.each do \|fd\|
	begin
	@buffers[fd] << fd.readpartial(4096)
	rescue EOFError
	@pool.delete fd
	end
	end
	end
	end

	def translate_output
	puts "Translating output..."
	@output = []
	@buffers.each do \|fd, buffer\|
	strio = StringIO.new(buffer)
	logs = []
	begin
	loop do
	logs << Marshal.load(strio)
	end
	rescue EOFError
	@output << logs
	end
	end
	end

	def analyze_output
	puts "Analyzing output..."
	call_times = []
	client_times = []
	connection_failures = []
	connection_errors = []
	shortest_call = 0
	shortest_client = 0
	longest_call = 0
	longest_client = 0
	@output.each do \|logs\|
	cur_call, cur_client = nil
	logs.each do \|tok, time\|
	case tok
	when :start
	cur_client = time
	when :call_start
	cur_call = time
	when :call_end
	delta = time - cur_call
	call_times << delta
	longest_call = delta unless longest_call > delta
	shortest_call = delta if shortest_call == 0 or delta < shortest_call
	cur_call = nil
	when :end
	delta = time - cur_client
	client_times << delta
	longest_client = delta unless longest_client > delta
	shortest_client = delta if shortest_client == 0 or delta < shortest_client
	cur_client = nil
	when :connection_failure
	connection_failures << time
	when :connection_error
	connection_errors << time
	end
	end
	end
	@report = {}
	@report[:total_calls] = call_times.inject(0.0) { \|a,t\| a += t }
	@report[:avg_calls] = @report[:total_calls] / call_times.size
	@report[:total_clients] = client_times.inject(0.0) { \|a,t\| a += t }
	@report[:avg_clients] = @report[:total_clients] / client_times.size
	@report[:connection_failures] = connection_failures.size
	@report[:connection_errors] = connection_errors.size
	@report[:shortest_call] = shortest_call
	@report[:shortest_client] = shortest_client
	@report[:longest_call] = longest_call
	@report[:longest_client] = longest_client
	@report[:total_benchmark_time] = @benchmark_end - @benchmark_start
	@report[:fastthread] = $".include?('fastthread.bundle')
	end

	def report_output
	fmt = "%.4f seconds"
	puts
	tabulate "%d",
	[["Server class", "%s"], @server.serverclass == Object ? "" : @server.serverclass],
	[["Server interpreter", "%s"], @server.interpreter],
	[["Client interpreter", "%s"], @interpreter],
	[["Socket class", "%s"], socket_class],
	["Number of processes", @num_processes],
	["Clients per process", @clients_per_process],
	["Calls per client", @calls_per_client],
	[["Using fastthread", "%s"], @report[:fastthread] ? "yes" : "no"]
	puts
	failures = (@report[:connection_failures] > 0)
	tabulate fmt,
	[["Connection failures", "%d", [:red, :bold]], @report[:connection_failures]],
	[["Connection errors", "%d", [:red, :bold]], @report[:connection_errors]],
	["Average time per call", @report[:avg_calls]],
	["Average time per client (%d calls)" % @calls_per_client, @report[:avg_clients]],
	["Total time for all calls", @report[:total_calls]],
	["Real time for benchmarking", @report[:total_benchmark_time]],
	["Shortest call time", @report[:shortest_call]],
	["Longest call time", @report[:longest_call]],
	["Shortest client time (%d calls)" % @calls_per_client, @report[:shortest_client]],
	["Longest client time (%d calls)" % @calls_per_client, @report[:longest_client]]
	end

	ANSI = {
	:reset => 0,
	:bold => 1,
	:black => 30,
	:red => 31,
	:green => 32,
	:yellow => 33,
	:blue => 34,
	:magenta => 35,
	:cyan => 36,
	:white => 37
	}

	def tabulate(fmt, *labels_and_values)
	labels = labels_and_values.map { \|l\| Array === l ? l.first : l }
	label_width = labels.inject(0) { \|w,l\| l.size > w ? l.size : w }
	labels_and_values.each do \|(l,v)\|
	f = fmt
	l, f, c = l if Array === l
	fmtstr = "%-#{label_width+1}s #{f}"
	if STDOUT.tty? and c and v.to_i > 0
	fmtstr = "\e[#{[c].map { \|x\| ANSI[x] } ";"}m" + fmtstr + "\e[#{ANSI[:reset]}m"
	end
	puts fmtstr % [l+":", v]
	end
	end
	end

	def resolve_const(const)
	const and const.split('::').inject(Object) { \|k,c\| k.const_get(c) }
	end

	puts "Starting server..."
	args = {}
	args[:interpreter] = ENV['THRIFT_SERVER_INTERPRETER'] \|\| ENV['THRIFT_INTERPRETER'] \|\| "ruby"
	args[:class] = resolve_const(ENV['THRIFT_SERVER']) \|\| Thrift::NonblockingServer
	args[:host] = ENV['THRIFT_HOST'] \|\| HOST
	args[:port] = (ENV['THRIFT_PORT'] \|\| PORT).to_i
	server = Server.new(args)
	server.start

	args = {}
	args[:host] = ENV['THRIFT_HOST'] \|\| HOST
	args[:port] = (ENV['THRIFT_PORT'] \|\| PORT).to_i
	args[:num_processes] = (ENV['THRIFT_NUM_PROCESSES'] \|\| 40).to_i
	args[:clients_per_process] = (ENV['THRIFT_NUM_CLIENTS'] \|\| 5).to_i
	args[:calls_per_client] = (ENV['THRIFT_NUM_CALLS'] \|\| 50).to_i
	args[:interpreter] = ENV['THRIFT_CLIENT_INTERPRETER'] \|\| ENV['THRIFT_INTERPRETER'] \|\| "ruby"
	args[:log_exceptions] = !!ENV['THRIFT_LOG_EXCEPTIONS']
	BenchmarkManager.new(args, server).run

	server.shutdown