src/swiftcore/Swiftiply/backend_protocol.rb src/swiftcore/Swiftiply/cluster_protocol.rb src/swiftcore/Swiftiply/constants.rb src/swiftcore/Swiftiply/proxy_bag.rb @@ -48,7 +48,7 @@ module Swiftcore require 'socket' require 'digest/sha2' require 'eventmachine' - require 'fastfilereaderext' + require 'fastfilereaderext' # This is unneeded as of EventMachine 0.12.6, so test for EM's version before requiring ours. require 'swiftcore/hash' require 'swiftcore/types' require 'swiftcore/Swiftiply/mocklog' @@ -61,14 +61,19 @@ module Swiftcore require 'swiftcore/splaytreemap' unless const_defined?(:HasSplayTree) HasSplayTree = true unless const_defined?(:HasSplayTree) - load_state = :remainder + load_state = :helpers require 'swiftcore/streamer' require 'swiftcore/Swiftiply/etag_cache' require 'swiftcore/Swiftiply/file_cache' require 'swiftcore/Swiftiply/dynamic_request_cache' require 'time' - # require 'swiftcore/microparser' - # require 'ruby-prof' + + load_state = :core + require 'swiftcore/Swiftiply/constants' + require 'swiftcore/Swiftiply/proxy_bag' + require 'swiftcore/Swiftiply/cluster_protocol' + require 'swiftcore/Swiftiply/backend_protocol' + rescue LoadError => e unless rubygems_loaded # Everything gets slower once rubygems are used (though this @@ -95,1143 +100,8 @@ module Swiftcore end GC.start - Deque = Array unless HasDeque or const_defined?(:Deque) module Swiftiply - Version = '0.6.4' - - # Yeah, these constants look kind of tacky. Inside of tight loops, - # though, using them makes a small but measurable difference, and those - # small differences add up.... - C_asterisk = '*'.freeze - C_empty = ''.freeze - C_header_close = 'HTTP/1.1 200 OK\r\nConnection: close\r\n'.freeze - C_header_keepalive = 'HTTP/1.1 200 OK\r\n'.freeze - C_slash = '/'.freeze - C_slashindex_html = '/index.html'.freeze - C1_0 = '1.0'.freeze - C1_1 = '1.1'.freeze - C_304 = "HTTP/1.1 304 Not Modified\r\n".freeze - Caos = 'application/octet-stream'.freeze - Cat = 'at'.freeze - Ccache_directory = 'cache_directory'.freeze - Ccache_extensions = 'cache_extensions'.freeze - Ccluster_address = 'cluster_address'.freeze - Ccluster_port = 'cluster_port'.freeze - Ccluster_server = 'cluster_server'.freeze - CConnection_close = "Connection: close\r\n".freeze - CConnection_KeepAlive = "Connection: Keep-Alive\r\n".freeze - CBackendAddress = 'BackendAddress'.freeze - CBackendPort = 'BackendPort'.freeze - Ccertfile = 'certfile'.freeze - Cchunked_encoding_threshold = 'chunked_encoding_threshold'.freeze - Cxforwardedfor = 'xforwardedfor'.freeze - Cdaemonize = 'daemonize'.freeze - Cdefault = 'default'.freeze - Cdescriptor_cache = 'descriptor_cache_threshold'.freeze - Cdescriptors = 'descriptors'.freeze - Cdocroot = 'docroot'.freeze - Cdynamic_request_cache = 'dynamic_request_cache'.freeze - Cenable_sendfile_404 = 'enable_sendfile_404'.freeze - Cepoll = 'epoll'.freeze - Cepoll_descriptors = 'epoll_descriptors'.freeze - Cetag_cache = 'etag_cache'.freeze - Cfile_cache = 'file_cache'.freeze - CGET = 'GET'.freeze - Cgroup = 'group'.freeze - CHEAD = 'HEAD'.freeze - Chost = 'host'.freeze - Cincoming = 'incoming'.freeze - Cinfo = 'info'.freeze - Ckeepalive = 'keepalive'.freeze - Ckey = 'key'.freeze - Ckeyfile = 'keyfile'.freeze - Cmap = 'map'.freeze - Cmax_cache_size = 'max_cache_size'.freeze - Cmsg_expired = 'browser connection expired'.freeze - Coutgoing = 'outgoing'.freeze - Cport = 'port'.freeze - Credeployable = 'redeployable'.freeze - Credeployment_sizelimit = 'redeployment_sizelimit'.freeze - Csendfileroot = 'sendfileroot'.freeze - Cservers = 'servers'.freeze - Cssl = 'ssl'.freeze - Csize = 'size'.freeze - Cstaticmask = 'staticmask'.freeze - Cswiftclient = 'swiftclient'.freeze - Cthreshold = 'threshold'.freeze - Ctimeslice = 'timeslice'.freeze - Ctimeout = 'timeout'.freeze - Curl = 'url'.freeze - Cuser = 'user'.freeze - Cwindow = 'window'.freeze - - C_fsep = File::SEPARATOR - - UnknownSocket = Socket::pack_sockaddr_in(0,'0.0.0.0') - - RunningConfig = {} - - class EMStartServerError < RuntimeError; end - class SwiftiplyLoggerNotFound < RuntimeError; end - - # The ProxyBag is a class that holds the client and the server queues, - # and that is responsible for managing them, matching them, and expiring - # them, if necessary. - - class ProxyBag - - attr_reader :keepalive_queue - - @client_q = Hash.new {|h,k| h[k] = Deque.new} - #@client_q = Hash.new {|h,k| h[k] = []} - @server_q = Hash.new {|h,k| h[k] = Deque.new} - #@server_q = Hash.new {|h,k| h[k] = []} - @keepalive_q = Deque.new - @logger = nil - @ctime = Time.now - @dateheader = "Date: #{@ctime.httpdate}\r\n\r\n" - @server_unavailable_timeout = 6 - @id_map = {} - @reverse_id_map = {} - @incoming_map = {} - @docroot_map = {} - @sendfileroot_map = {} - @log_map = {} - @redeployable_map = {} - @file_cache_map = {} - @dynamic_request_map = {} - @etag_cache_map = {} - @x_forwarded_for = {} - @keepalive = {} - @static_mask = {} - @keys = {} - @filters = {} - @demanding_clients = Hash.new {|h,k| h[k] = Deque.new} - #@demanding_clients = Hash.new {|h,k| h[k] = []} - @hitcounters = Hash.new {|h,k| h[k] = 0} - # Kids, don't do this at home. It's gross. - @typer = MIME::Types.instance_variable_get('@__types__') - - MockLog = Swiftcore::Swiftiply::MockLog.new - - class << self - - def now - @ctime - end - - # Setter and Getter accessors for the logger. - def logger=(val) - @logger = val - end - - def logger - @logger - end - - def log_level=(val) - @log_level = val - end - - def log_level - @log_level - end - - # Returns the access key. If an access key is set, then all new - # backend connections must send the correct access key before - # being added to the cluster as a valid backend. - - def get_key(h) - @keys[h] || C_empty - end - - def set_key(h,val) - @keys[h] = val - end - - def add_id(who,what) - @id_map[who] = what - @reverse_id_map[what] = who - end - - def remove_id(who) - what = @id_map.delete(who) - @reverse_id_map.delete(what) - end - - def incoming_mapping(name) - @incoming_map[name] - end - - def add_incoming_mapping(hashcode,name) - @incoming_map[name] = hashcode - end - - def remove_incoming_mapping(name) - @incoming_map.delete(name) - end - - def add_docroot(path,name) - @docroot_map[name] = File.expand_path(path) - end - - def remove_docroot(name) - @docroot_map.delete(name) - end - - def add_sendfileroot(path,name) - @sendfileroot_map[name] = path - end - - def remove_sendfileroot(name) - @sendfileroot_map.delete(name) - end - - def get_sendfileroot(name) - @sendfileroot_map[name] - end - - def add_redeployable(limit,name) - @redeployable_map[name] = limit - end - - def remove_redeployable(name) - @redeployable_map.delete(name) - end - - def add_log(log,name) - @log_map[name] = [log,1] - end - - def log(name) - (@log_map[name] && @log_map[name].first) || MockLog - end - - def remove_log(name) - @log_map[name].close if @log_map[name].respond_to? :close - @log_map.delete(name) - end - - def set_level(level,name) - @log_map[name][1] = level - end - - def level(name) - (@log_map[name] && @log_map[name].last) || 0 - end - - def add_file_cache(cache,name) - @file_cache_map[name] = cache - end - - def add_dynamic_request_cache(cache,name) - @dynamic_request_map[name] = cache - end - - def add_etag_cache(cache,name) - @etag_cache_map[name] = cache - end - - def x_forwarded_for(name) - @x_forwarded_for[name] - end - - def set_x_forwarded_for(name) - @x_forwarded_for[name] = true - end - - def unset_x_forwarded_for(name) - @x_forwarded_for[name] = false - end - - def add_static_mask(regexp, name) - @static_mask[name] = regexp - end - - def static_mask(name) - @static_mask[name] - end - - def remove_static_mask(name) - @static_mask.delete(name) - end - - def add_filter(filter, name) - (@filters[name] ||= []) << filter - end - - def filter(name) - @filters[name] - end - - def remove_filters(name) - @filters[name].clear if @filters[name] - end - - def add_keepalive(timeout, name) - @keepalive[name] = timeout == 0 ? false : timeout - end - - def keepalive(name) - @keepalive[name] - end - - def remove_keepalive(name) - @keepalive[name] = false - end - - # Sets the default proxy destination, if requests are received - # which do not match a defined destination. - - def default_name - @default_name - end - - def default_name=(val) - @default_name = val - end - - # This timeout is the amount of time a connection will sit in queue - # waiting for a backend to process it. A client connection that - # sits for longer than this timeout receives a 503 response and - # is dropped. - - def server_unavailable_timeout - @server_unavailable_timeout - end - - def server_unavailable_timeout=(val) - @server_unavailable_timeout = val - end - - # The chunked_encoding_threshold is a file size limit. Files - # which fall below this limit are sent in one chunk of data. - # Files which hit or exceed this limit are delivered via chunked - # encoding. This enforces a maximum threshold of 32k. - - def chunked_encoding_threshold - @chunked_enconding_threshold || 32768 - end - - def chunked_encoding_threshold=(val) - @chunked_encoding_threshold = val > 32768 ? 32768 : val - end - - def cache_threshold - @cache_threshold || 32768 - end - - def cache_threshold=(val) - @cache_threshold = val > 256*1024 ? 256*1024 : val - end - - # Swiftiply maintains caches of small static files, etags, and - # dynamnic request paths for each cluster of backends. - # A timer is created when each cache is created, to do the - # initial update. Thereafer, the verification method on the - # cache returns the number of seconds to wait before running - # again. - - def verify_cache(cache) - log(cache.owner_hash).log(Cinfo,"Checking #{cache.class.name}(#{cache.vqlength}/#{cache.length}) for #{cache.owners}") if level(cache.owner_hash) > 2 - new_interval = cache.check_verification_queue - log(cache.owner_hash).log(Cinfo," Next #{cache.class.name} check in #{new_interval} seconds") if level(cache.owner_hash) > 2 - EventMachine.add_timer(new_interval) do - verify_cache(cache) - end - end - - # Handle static files. It employs an extension to efficiently - # handle large files, and depends on an addition to - # EventMachine, send_file_data(), to efficiently handle small - # files. In my tests, it streams in excess of 120 megabytes of - # data per second for large files, and does 8000+ to 9000+ - # requests per second with small files (i.e. under 4k). I think - # this can still be improved upon for small files. - # - # This code is damn ugly. - - def serve_static_file(clnt,dr = nil) - request_method = clnt.request_method - - # Only GET and HEAD requests can return a file. - if request_method == CGET || request_method == CHEAD - path_info = clnt.uri - client_name = clnt.name - dr ||= @docroot_map[client_name] - fc = @file_cache_map[client_name] - - if data = fc[path_info] - none_match = clnt.none_match - same_response = case - when request_method == CHEAD then false - when none_match && none_match == C_asterisk then false - when none_match && !none_match.strip.split(/\s*,\s*/).include?(data[1]) then false - else none_match - end - if same_response - clnt.send_data "#{C_304}#{clnt.connection_header}Content-Length: 0\r\n#{@dateheader}" - oh = fc.owner_hash - log(oh).log(Cinfo,"#{Socket::unpack_sockaddr_in(clnt.get_peername || UnknownSocket).last} \"GET #{path_info} HTTP/#{clnt.http_version}\" 304 -") if level(oh) > 1 - else - unless request_method == CHEAD - clnt.send_data "#{data.last}#{clnt.connection_header}#{@dateheader}#{data.first}" - oh = fc.owner_hash - log(oh).log(Cinfo,"#{Socket::unpack_sockaddr_in(clnt.get_peername || UnknownSocket).last} \"GET #{path_info} HTTP/#{clnt.http_version}\" 200 #{data.first.length}") if level(oh) > 1 - else - clnt.send_data "#{data.last}#{clnt.connection_header}#{@dateheader}" - oh = fc.owner_hash - log(oh).log(Cinfo,"#{Socket::unpack_sockaddr_in(clnt.get_peername || UnknownSocket).last} \"HEAD #{path_info} HTTP/#{clnt.http_version}\" 200 -") if level(oh) > 1 - end - end - - unless clnt.keepalive - clnt.close_connection_after_writing - else - clnt.reset_state - end - - true - elsif path = find_static_file(dr,path_info,client_name) # See if the static file that is wanted exists on the filesystem. - #TODO: There is a race condition here between when we detect whether the file is there, and when we start to deliver it. - # It'd be nice to handle an exception when trying to read the file in a graceful way, by falling out as if no static - # file had been found. That way, if the file is deleted between detection and the start of delivery, such as might - # happen when delivering files out of some sort of page cache, it can be handled in a reasonable manner. This should - # be easily doable, so DO IT SOON! - none_match = clnt.none_match - etag,mtime = @etag_cache_map[client_name].etag_mtime(path) - same_response = nil - same_response = case - when request_method == CHEAD then false - when none_match && none_match == C_asterisk then false - when none_match && !none_match.strip.split(/\s*,\s*/).include?(etag) then false - else none_match - end - - if same_response - clnt.send_data "#{C_304}#{clnt.connection_header}Content-Length: 0\r\n#{@dateheader}" - - unless clnt.keepalive - clnt.close_connection_after_writing - else - clnt.reset_state - end - - oh = fc.owner_hash - log(oh).log(Cinfo,"#{Socket::unpack_sockaddr_in(clnt.get_peername || UnknownSocket).last} \"GET #{path_info} HTTP/#{clnt.http_version}\" 304 -") if level(oh) > 1 - else - ct = @typer.simple_type_for(path) || Caos - fsize = File.size(path) - - header_line = "HTTP/1.1 200 OK\r\nETag: #{etag}\r\nContent-Type: #{ct}\r\nContent-Length: #{fsize}\r\n" - - fd = nil - if fsize < @chunked_encoding_threshold - File.open(path) {|fh| fd = fh.sysread(fsize)} - clnt.send_data "#{header_line}#{clnt.connection_header}#{@dateheader}" - unless request_method == CHEAD - if fsize < 32768 - clnt.send_file_data path - else - clnt.send_data fd - end - end - - unless clnt.keepalive - clnt.close_connection_after_writing - else - clnt.reset_state - end - - elsif clnt.http_version != C1_0 && fsize > @chunked_encoding_threshold - clnt.send_data "HTTP/1.1 200 OK\r\n#{clnt.connection_header}ETag: #{etag}\r\nContent-Type: #{ct}\r\nTransfer-Encoding: chunked\r\n#{@dateheader}" - EM::Deferrable.future(clnt.stream_file_data(path, :http_chunks=>true)) {clnt.close_connection_after_writing} unless request_method == CHEAD - else - #clnt.send_data "HTTP/1.1 200 OK\r\nConnection: close\r\nETag: #{etag}\r\nContent-Type: #{ct}\r\nContent-Length: #{fsize}\r\n\r\n" - clnt.send_data "#{header_line}#{clnt.connection_header}#{@dateheader}" - EM::Deferrable.future(clnt.stream_file_data(path, :http_chunks=>false)) {clnt.close_connection_after_writing} unless request_method == CHEAD - end - - fc.add(path_info, path, fd || File.read(path),etag,mtime,header_line) if fsize < @cache_threshold - - oh = fc.owner_hash - log(oh).log(Cinfo,"#{Socket::unpack_sockaddr_in(clnt.get_peername || UnknownSocket).last} \"#{request_method} #{path_info} HTTP/#{clnt.http_version}\" 200 #{request_method == CHEAD ? C_empty : fsize}") if level(oh) > 1 - end - true - end - else - false - end - # The exception is going to be eaten here, because some - # dumb file IO error shouldn't take Swiftiply down. - rescue Object => e - puts e - @logger.log('error',"Failed request for #{dr.inspect}/#{path.inspect} -- #{e} @ #{e.backtrace.inspect}") if @log_level > 0 - - # TODO: This is uncivilized; if there is an unexpected error, a reasonable response MUST be returned. - clnt.close_connection_after_writing - false - end - - # Determine if the requested file, in the given docroot, exists - # and is a file (i.e. not a directory). - # - # If Rails style page caching is enabled, this method will be - # dynamically replaced by a more sophisticated version. - - def find_static_file(docroot,path_info,client_name) - path = File.join(docroot,path_info) - path if FileTest.exist?(path) and FileTest.file?(path) and File.expand_path(path).index(docroot) == 0 and !(x = static_mask(client_name) and path =~ x) ? path : false - end - - # Pushes a front end client (web browser) into the queue of - # clients waiting to be serviced if there's no server available - # to handle it right now. - - def add_frontend_client(clnt,data_q,data) - clnt.create_time = @ctime - - # Initialize parameters relevant to redeployable requests, if this client - # has them enabled. - clnt.data_pos = clnt.data_len = 0 if clnt.redeployable = @redeployable_map[clnt.name] - - uri = clnt.uri - name = clnt.name - drm = @dynamic_request_map[name] - if drm[uri] || !(@docroot_map.has_key?(name) && serve_static_file(clnt)) - # It takes two requests to add it to the dynamic verification - # queue. So, go from nil to false, then from false to - # insertion into the queue. - unless drmval = drm[uri] - if drmval == false - drm[uri] = drm.add_to_verification_queue(uri) - log(drm.owner_hash).log(Cinfo,"Adding request #{uri} to dynamic request cache") if level(drm.owner_hash) > 2 - else - drm[uri] = false - end - end - - # A lot of sites won't need to check X-FORWARDED-FOR, so - # we'll only take the time to munge the headers to add - # it if the config calls for it. - if x_forwarded_for(clnt.name) and peername = clnt.get_peername - data.sub!(/\r\n\r\n/,"\r\nX-FORWARDED-FOR: #{Socket::unpack_sockaddr_in(peername).last}\r\n\r\n") - end - - data_q.unshift data - unless match_client_to_server_now(clnt) - if clnt.uri =~ /\w+-\w+-\w+\.\w+\.[\w\.]+-(\w+)?$/ - @demanding_clients[$1].unshift clnt - else - @client_q[@incoming_map[name]].unshift(clnt) - end - end - end - end - - def rebind_frontend_client(clnt) - clnt.create_time = @ctime - clnt.data_pos = clnt.data_len = 0 - - unless match_client_to_server_now(clnt) - if clnt.uri =~ /\w+-\w+-\w+\.\w+\.[\w\.]+-(\w+)?$/ - #if $& #### - @demanding_clients[$1].unshift clnt - else - @client_q[@incoming_map[clnt.name]].unshift(clnt) - end - end - end - - # Pushes a backend server into the queue of servers waiting for - # a client to service if there are no clients waiting to be - # serviced. - - def add_server srvr - if f = srvr.filter - q[f] = Deque.new unless q = @server_q[srvr.name] - q[f].unshift(srvr) unless match_server_to_client_now(srvr) - else - @server_q[srvr.name].unshift(srvr) unless match_server_to_client_now(srvr) - end - end - - # Deletes the provided server from the server queue. - - def remove_server srvr - @server_q[srvr.name].delete srvr - end - - # Removes the named client from the client queue. - # TODO: Try replacing this with ...something. Performance - # here has to be bad when the list is long. - - def remove_client clnt - @client_q[clnt.name].delete clnt - end - - # Tries to match the client (passed as an argument) to a - # server. - - def match_client_to_server_now(client) - hash = @incoming_map[client.name] - - if outgoing_filters = @filters[hash] - outgoing_filters.each do |f| - if client.uri =~ f - sq = @server_q[@incoming_map[client.name][f]] - break - end - end - end - - sq ||= @server_q[@incoming_map[client.name]] - if sq.empty? - false - elsif client.uri =~ /\w+-\w+-\w+\.\w+\.[\w\.]+-(\w+)?$/ - if sidx = sq.index(@reverse_id_map[$1]) - server = sq.delete_at(sidx) - server.associate = client - client.associate = server - client.push - true - else - false - end - elsif server = sq.pop - server.associate = client - client.associate = server - client.push - true - else - false - end - end - - # Tries to match the server (passed as an argument) to a - # client. - - def match_server_to_client_now(server) - if client = @demanding_clients[server.id].pop - server.associate = client - client.associate = server - client.push - true - elsif client = @client_q[server.name].pop - server.associate = client - client.associate = server - client.push - true - else - false - end - end - - # Walk through the waiting clients if there is no server - # available to process clients and expire any clients that - # have been waiting longer than @server_unavailable_timeout - # seconds. Clients which are expired will receive a 503 - # response. If this is happening, either you need more - # backend processes, or your @server_unavailable_timeout is - # too short. - - def expire_clients - now = Time.now - @client_q.each_key do |name| - while c = @client_q[name].pop - if (now - c.create_time) >= @server_unavailable_timeout - c.send_503_response - else - @client_q[name].push c - break - end - end - end - end - - # This is called by a periodic timer once a second to update - # the time. - - def update_ctime - @ctime = Time.now - @dateheader = "Date: #{@ctime.httpdate}\r\n\r\n" - end - - end - end - - # The ClusterProtocol is the subclass of EventMachine::Connection used - # to communicate between Swiftiply and the web browser clients. - - class ClusterProtocol < EventMachine::Connection - -# include Swiftcore::MicroParser - - attr_accessor :create_time, :last_action_time, :uri, :associate, :name, :redeployable, :data_pos, :data_len, :peer_ip, :connection_header, :keepalive - - Crn = "\r\n".freeze - Crnrn = "\r\n\r\n".freeze - C_blank = ''.freeze - C_percent = '%'.freeze - Cunknown_host = 'unknown host'.freeze - C503Header = "HTTP/1.1 503 Server Unavailable\r\nContent-Type: text/plain\r\nConnection: close\r\n\r\n" - C404Header = "HTTP/1.1 404 Not Found\r\nContent-Type: text/plain\r\nConnection: close\r\n\r\n" - C400Header = "HTTP/1.1 400 Bad Request\r\nContent-Type: text/plain\r\nConnection: close\r\n\r\n" - @count_connections = 0 - @count_400s = 0 - @count_404s = 0 - @count_503s = 0 - - # Initialize the @data array, which is the temporary storage for blocks - # of data received from the web browser client, then invoke the superclass - # initialization. - - def initialize *args - @data = Deque.new - #@data = [] - @data_pos = 0 - @connection_header = C_empty - @hmp = @name = @uri = @http_version = @request_method = @none_match = @done_parsing = nil - @keepalive = true - super - end - - def reset_state - @data.clear - @data_pos = 0 - @connection_header = C_empty - @hmp = @name = @uri = @http_version = @request_method = @none_match = @done_parsing = nil - @keepalive = true - end - - # States: - # uri - # name - # \r\n\r\n - # If-None-Match - # Done Parsing - def receive_data data - if @done_parsing - @data.unshift data - push - else - unless @uri - # It's amazing how, when writing the code, the brain can be in a zone - # where line noise like this regexp makes perfect sense, and is clear - # as day; one looks at it and it reads like a sentence. Then, one - # comes back to it later, and looks at it when the brain is in a - # different zone, and 'lo! It looks like line noise again. - # - # data =~ /^(\w+) +(?:\w+:\/\/([^\/]+))?([^ \?]+)\S* +HTTP\/(\d\.\d)/ - # - # In case it looks like line noise to you, dear reader, too: - # - # 1) Match and save the first set of word characters. - # - # Followed by one or more spaces. - # - # Match but do not save the word characters followed by :// - # - # 2) Match and save one or more characters that are not a slash - # - # And allow this whole thing to match 1 or 0 times. - # - # 3) Match and save one or more characters that are not a question - # mark or a space. - # - # Match zero or more non-whitespace characters, followed by one - # or more spaces, followed by "HTTP/". - # - # 4) Match and save a digit dot digit. - # - # Thus, this pattern will match both the standard: - # GET /bar HTTP/1.1 - # style request, as well as the valid (for a proxy) but less common: - # GET http://foo/bar HTTP/1.0 - # - # If the match fails, then this is a bad request, and an appropriate - # response will be returned. - # - # http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec5.1.2 - # - if data =~ /^(\w+) +(?:\w+:\/\/([^ \/]+))?([^ \?\#]*)\S* +HTTP\/(\d\.\d)/ - @request_method = $1 - @uri = $3 - @http_version = $4 - if $2 - @name = $2.intern - @uri = C_slash if @uri.empty? - # Rewrite the request to get rid of the http://foo portion. - - data.sub!(/^\w+ +\w+:\/\/[^ \/]+([^ \?]*)/,"#{@request_method} #{@uri}") - end - @uri = @uri.tr('+', ' ').gsub(/((?:%[0-9a-fA-F]{2})+)/n) {[$1.delete(C_percent)].pack('H*')} if @uri.include?(C_percent) - else - send_400_response - return - end - end - unless @name - if data =~ /^Host: *([^\r\0:]+)/ - @name = $1.intern - end - end - if @hmp - # Hopefully this doesn't happen often. - d = @data.to_s - else - d = data - @hmp = true - end - if d.include?(Crnrn) - @name = ProxyBag.default_name unless ProxyBag.incoming_mapping(@name) - @done_parsing = true - if data =~ /If-None-Match: *([^\r]+)/ - @none_match = $1 - end - - # Keep-Alive works differently on HTTP 1.0 versus HTTP 1.1 - # HTTP 1.0 was not written to support Keep-Alive initially; it was - # bolted on. Thus, for an HTTP 1.0 client to indicate that it - # wants to initiate a Keep-Alive session, it must send a header: - # - # Connection: Keep-Alive - # - # Then, when the server sends the response, it must likewise add: - # - # Connection: Keep-Alive - # - # to the response. - # - # For HTTP 1.1, Keep-Alive is assumed. If a client does not want - # Keep-Alive, then it must send the following header: - # - # Connection: close - # - # Likewise, if the server does not want to keep the connection - # alive, it must send the same header: - # - # Connection: close - # - # to the client. - - if @name - unless ProxyBag.keepalive(@name) == false - if @http_version == C1_0 - if data =~ /Connection: Keep-Alive/i - # Nonstandard HTTP 1.0 situation; apply keepalive header. - @connection_header = CConnection_KeepAlive - else - # Standard HTTP 1.0 situation; connection will be closed. - @keepalive = false - @connection_header = CConnection_close - end - else # The connection is an HTTP 1.1 connection. - if data =~ /Connection: [Cc]lose/ - # Nonstandard HTTP 1.1 situation; connection will be closed. - @keepalive = false - end - end - end - - ProxyBag.add_frontend_client(self,@data,data) - else - send_404_response - end - else - @data.unshift data - end - end - end - - # Hardcoded 400 response that is sent if the request is malformed. - - def send_400_response - ip = Socket::unpack_sockaddr_in(get_peername).last rescue Cunknown_host - error = "The request received on #{ProxyBag.now.asctime} from #{ip} was malformed and could not be serviced." - send_data "#{C400Header}Bad Request\n\n#{error}" - ProxyBag.logger.log(Cinfo,"Bad Request -- #{error}") - close_connection_after_writing - end - - # Hardcoded 404 response. This is sent if a request can't be matched to - # any defined incoming section. - - def send_404_response - ip = Socket::unpack_sockaddr_in(get_peername).last rescue Cunknown_host - error = "The request (#{@uri} --> #{@name}), received on #{ProxyBag.now.asctime} from #{ip} did not match any resource know to this server." - send_data "#{C404Header}Resource not found.\n\n#{error}" - ProxyBag.logger.log(Cinfo,"Resource not found -- #{error}") - close_connection_after_writing - end - - # Hardcoded 503 response that is sent if a connection is timed out while - # waiting for a backend to handle it. - - def send_503_response - ip = Socket::unpack_sockaddr_in(get_peername).last rescue Cunknown_host - error = "The request (#{@uri} --> #{@name}), received on #{create_time.asctime} from #{ip} timed out before being deployed to a server for processing." - send_data "#{C503Header}Server Unavailable\n\n#{error}" - ProxyBag.logger.log(Cinfo,"Server Unavailable -- #{error}") - close_connection_after_writing - end - - # Push data from the web browser client to the backend server process. - - def push - if @associate - unless @redeployable - # normal data push - data = nil - @associate.send_data data while data = @data.pop - else - # redeployable data push; just send the stuff that has - # not already been sent. - (@data.length - 1 - @data_pos).downto(0) do |p| - d = @data[p] - @associate.send_data d - @data_len += d.length - end - @data_pos = @data.length - - # If the request size crosses the size limit, then - # disallow redeployent of this request. - if @data_len > @redeployable - @redeployable = false - @data.clear - end - end - end - end - - # The connection with the web browser client has been closed, so the - # object must be removed from the ProxyBag's queue if it is has not - # been associated with a backend. If it has already been associated - # with a backend, then it will not be in the queue and need not be - # removed. - - def unbind - ProxyBag.remove_client(self) unless @associate - end - - def request_method; @request_method; end - def http_version; @http_version; end - def none_match; @none_match; end - - def setup_for_redeployment - @data_pos = 0 - end - - end - - # The BackendProtocol is the EventMachine::Connection subclass that - # handles the communications between Swiftiply and the backend process - # it is proxying to. - - class BackendProtocol < EventMachine::Connection - attr_accessor :associate, :id - - C0rnrn = "0\r\n\r\n".freeze - Crnrn = "\r\n\r\n".freeze - - def initialize *args - @name = self.class.bname - @permit_xsendfile = self.class.xsendfile - @enable_sendfile_404 = self.class.enable_sendfile_404 - super - end - - def name - @name - end - - # Call setup() and add the backend to the ProxyBag queue. - - def post_init - setup - @initialized = nil - ProxyBag.add_server self - end - - # Setup the initial variables for receiving headers and content. - - def setup - @headers = '' - @headers_completed = @dont_send_data = false - #@content_length = nil - @content_sent = 0 - @filter = self.class.filter - end - - # Receive data from the backend process. Headers are parsed from - # the rest of the content. If a Content-Length header is present, - # that is used to determine how much data to expect. Otherwise, - # if 'Transfer-encoding: chunked' is present, assume chunked - # encoding. Otherwise be paranoid; something isn't the way we like - # it to be. - - def receive_data data - unless @initialized - # preamble = data.slice!(0..24) - preamble = data[0..24] - data = data[25..-1] || C_empty - keylen = preamble[23..24].to_i(16) - keylen = 0 if keylen < 0 - key = keylen > 0 ? data.slice!(0..(keylen - 1)) : C_empty - #if preamble[0..10] == Cswiftclient and key == ProxyBag.get_key(@name) - if preamble.index(Cswiftclient) == 0 and key == ProxyBag.get_key(@name) - @id = preamble[11..22] - ProxyBag.add_id(self,@id) - @initialized = true - else - # The worker that connected did not present the proper authentication, - # so something is fishy; time to cut bait. - close_connection - return - end - end - - unless @headers_completed - if data.include?(Crnrn) - @headers_completed = true - h,data = data.split(/\r\n\r\n/,2) - #@headers << h << Crnrn - if @headers.length > 0 - @headers << h - else - @headers = h - end - - if @headers =~ /Content-[Ll]ength: *([^\r]+)/ - @content_length = $1.to_i - elsif @headers =~ /Transfer-encoding:\s*chunked/ - @content_length = nil - else - @content_length = 0 - end - - if @permit_xsendfile && @headers =~ /X-[Ss]endfile: *([^\r]+)/ - @associate.uri = $1 - if ProxyBag.serve_static_file(@associate,ProxyBag.get_sendfileroot(@associate.name)) - @dont_send_data = true - else - if @enable_sendfile_404 - msg = "#{@associate.uri} could not be found." - @associate.send_data "HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\nContent-Type: text/html\r\nContent-Length: #{msg.length}\r\n\r\n#{msg}" - @associate.close_connection_after_writing - @dont_send_data = true - else - @associate.send_data @headers + Crnrn - end - end - else - @associate.send_data @headers + Crnrn - end - - # If keepalive is turned on, the assumption is that it will stay - # on, unless the headers being returned indicate that the connection - # should be closed. - # So, check for a 'Connection: Closed' header. - if keepalive = @associate.keepalive - keepalive = false if @headers =~ /Connection: [Cc]lose/ - if @associate_http_version == C1_0 - keepalive = false unless @headers == /Connection: Keep-Alive/i - end - end - else - @headers << data - end - end - - if @headers_completed - @associate.send_data data unless @dont_send_data - @content_sent += data.length - if @content_length and @content_sent >= @content_length or data[-6..-1] == C0rnrn - # If @dont_send_data is set, then the connection is going to be closed elsewhere. - unless @dont_send_data - # Check to see if keepalive is enabled. - if keepalive - @associate.reset_state - ProxyBag.remove_client(self) unless @associate - else - @associate.close_connection_after_writing - end - end - @associate = @headers_completed = @dont_send_data = nil - @headers = '' - #@headers_completed = false - #@content_length = nil - @content_sent = 0 - #setup - ProxyBag.add_server self - end - end - # TODO: Log these errors! - rescue Exception => e - puts "Kaboom: #{e} -- #{e.backtrace.inspect}" - @associate.close_connection_after_writing if @associate - @associate = nil - setup - ProxyBag.add_server self - end - - # This is called when the backend disconnects from the proxy. - # If the backend is currently associated with a web browser client, - # that connection will be closed. Otherwise, the backend will be - # removed from the ProxyBag's backend queue. - - def unbind - if @associate - if !@associate.redeployable or @content_length - @associate.close_connection_after_writing - else - @associate.associate = nil - @associate.setup_for_redeployment - ProxyBag.rebind_frontend_client(@associate) - end - else - ProxyBag.remove_server(self) - end - ProxyBag.remove_id(self) - end - - def self.bname=(val) - @bname = val - end - - def self.bname - @bname - end - - def self.xsendfile=(val) - @xsendfile = val - end - - def self.xsendfile - @xsendfile - end - - def self.enable_sendfile_404=(val) - @enable_sendfile_404 = val - end - - def self.enable_sendfile_404 - @enable_sendfile_404 - end - - def self.filter=(val) - @filter = val - end - - def self.filter - @filter - end - - def filter - @filter - end - end # Start the EventMachine event loop and create the front end and backend # handlers, then create the timers that are used to expire unserviced @@ -1241,10 +111,9 @@ module Swiftcore @existing_backends = {} # Default is to assume we want to try to turn epoll/kqueue support on. - EventMachine.kqueue - #EventMachine.epoll unless config.has_key?(Cepoll) and !config[Cepoll] rescue nil - #EventMachine.kqueue unless config.has_key?(Ckqueue) and !config[Ckqueue] rescue nil - #EventMachine.set_descriptor_table_size(config[Cepoll_descriptors] || config[Cdescriptors] || 4096) rescue nil + EventMachine.epoll unless config.has_key?(Cepoll) and !config[Cepoll] rescue nil + EventMachine.kqueue unless config.has_key?(Ckqueue) and !config[Ckqueue] rescue nil + EventMachine.set_descriptor_table_size(config[Cepoll_descriptors] || config[Cdescriptors] || 4096) rescue nil EventMachine.run do EM.set_timer_quantum(5) @@ -1307,11 +176,13 @@ EOC port, ssl_protocol) else + standard_protocol = Class.new(ClusterProtocol) + standard_protocol.init_class_variables ProxyBag.logger.log(Cinfo,"Opening server on #{address}:#{port}") if ProxyBag.log_level > 0 new_config[Ccluster_server][addrport] = EventMachine.start_server( address, port, - ClusterProtocol) + standard_protocol) end rescue RuntimeError => e advice = '' src/swiftcore/Swiftiply.rb @@ -144,6 +144,7 @@ module Mongrel #@acceptor = Thread.new do @acceptor = Thread.current EventMachine.run do + EM.set_timer_quantum(5) begin MongrelProtocol.connect(@host,@port.to_i,@key) rescue StopServer src/swiftcore/swiftiplied_mongrel.rb @@ -245,11 +245,11 @@ ECONF # Hit it a bunch of times. ab = `which ab`.chomp - unless ab == '' - r = `#{ab} -n 100000 -c 25 http://127.0.0.1:29998/#{smallfile_name}` - r =~ /^(Requests per second.*)$/ - puts "10k 1020 byte files, concurrency of 25; no KeepAlive\n#{$1}\n" - end + #unless ab == '' + # r = `#{ab} -n 100000 -c 2 http://127.0.0.1:29998/#{smallfile_name}` + # r =~ /^(Requests per second.*)$/ + # puts "10k 1020 byte files, concurrency of 25; no KeepAlive\n#{$1}\n" + #end unless ab == '' r = `#{ab} -n 100000 -c 25 -k http://127.0.0.1:29998/#{smallfile_name}` @@ -263,7 +263,7 @@ ECONF puts "10k 1020 byte files with etag, concurrency of 25; with KeepAlive\n#{$1}\n" end unless ab == '' - r = `#{ab} -n 100000 -i -c 25 http://127.0.0.1:29998/#{smallfile_name}` + r = `#{ab} -n 10000 -i -c 25 http://127.0.0.1:29998/#{smallfile_name}` r =~ /^(Requests per second.*)$/ puts "10k HEAD requests, concurrency of 25\n#{$1}\n" end @@ -884,7 +884,7 @@ ECONF ab = `which ab`.chomp unless ab == '' - r = `#{ab} -n 100000 -c 25 http://127.0.0.1:29998/hello` + r = `#{ab} -n 10000 -c 25 http://127.0.0.1:29998/hello` r =~ /^(Requests per second.*)$/ puts "Swiftiply -> Swiftiplied Mongrel, concurrency of 25\n#{$1}" end test/TC_Swiftiply.rb 8cc0a8a4be53aadee6fc07cf1f4fc07e7cbaec28 Kirk Haines khaines@engineyard.com http://github.com/wyhaines/swiftiply/commit/064f6e184f995eb909cd13a31b3b6591ca874bb9 064f6e184f995eb909cd13a31b3b6591ca874bb9 2009-04-10T10:31:57-07:00 2009-04-10T10:31:57-07:00 Started refactoring the massive Swiftiply.rb into separate files for various pieces. Added a bit of code for the status page. Tweaked things crudely to get around buggy 'ab' on OSX when running the integration tests in the test suite. 31d4a6129f1197f35695dda8bc1669200b82d6c9 Kirk Haines khaines@engineyard.com