#
# 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 'thrift/server/nonblockingserver'
require 'thrift/transport/unixsocket'
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