+++ /dev/null
-#
-# 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 'logger'
-require 'thread'
-
-module Thrift
- # this class expects to always use a FramedTransport for reading messages
- class NonblockingServer < BaseServer
- def initialize(processor, server_transport, transport_factory=nil, protocol_factory=nil, num=20, logger=nil)
- super(processor, server_transport, transport_factory, protocol_factory)
- @num_threads = num
- if logger.nil?
- @logger = Logger.new(STDERR)
- @logger.level = Logger::WARN
- else
- @logger = logger
- end
- @shutdown_semaphore = Mutex.new
- @transport_semaphore = Mutex.new
- end
-
- def serve
- @logger.info "Starting #{self}"
- @server_transport.listen
- @io_manager = start_io_manager
-
- begin
- loop do
- break if @server_transport.closed?
- begin
- rd, = select([@server_transport], nil, nil, 0.1)
- rescue Errno::EBADF => e
- # In Ruby 1.9, calling @server_transport.close in shutdown paths causes the select() to raise an
- # Errno::EBADF. If this happens, ignore it and retry the loop.
- break
- end
- next if rd.nil?
- socket = @server_transport.accept
- @logger.debug "Accepted socket: #{socket.inspect}"
- @io_manager.add_connection socket
- end
- rescue IOError => e
- end
- # we must be shutting down
- @logger.info "#{self} is shutting down, goodbye"
- ensure
- @transport_semaphore.synchronize do
- @server_transport.close
- end
- @io_manager.ensure_closed unless @io_manager.nil?
- end
-
- def shutdown(timeout = 0, block = true)
- @shutdown_semaphore.synchronize do
- return if @is_shutdown
- @is_shutdown = true
- end
- # nonblocking is intended for calling from within a Handler
- # but we can't change the order of operations here, so lets thread
- shutdown_proc = lambda do
- @io_manager.shutdown(timeout)
- @transport_semaphore.synchronize do
- @server_transport.close # this will break the accept loop
- end
- end
- if block
- shutdown_proc.call
- else
- Thread.new &shutdown_proc
- end
- end
-
- private
-
- def start_io_manager
- iom = IOManager.new(@processor, @server_transport, @transport_factory, @protocol_factory, @num_threads, @logger)
- iom.spawn
- iom
- end
-
- class IOManager # :nodoc:
- DEFAULT_BUFFER = 2**20
-
- def initialize(processor, server_transport, transport_factory, protocol_factory, num, logger)
- @processor = processor
- @server_transport = server_transport
- @transport_factory = transport_factory
- @protocol_factory = protocol_factory
- @num_threads = num
- @logger = logger
- @connections = []
- @buffers = Hash.new { |h,k| h[k] = '' }
- @signal_queue = Queue.new
- @signal_pipes = IO.pipe
- @signal_pipes[1].sync = true
- @worker_queue = Queue.new
- @shutdown_queue = Queue.new
- end
-
- def add_connection(socket)
- signal [:connection, socket]
- end
-
- def spawn
- @iom_thread = Thread.new do
- @logger.debug "Starting #{self}"
- run
- end
- end
-
- def shutdown(timeout = 0)
- @logger.debug "#{self} is shutting down workers"
- @worker_queue.clear
- @num_threads.times { @worker_queue.push [:shutdown] }
- signal [:shutdown, timeout]
- @shutdown_queue.pop
- @signal_pipes[0].close
- @signal_pipes[1].close
- @logger.debug "#{self} is shutting down, goodbye"
- end
-
- def ensure_closed
- kill_worker_threads if @worker_threads
- @iom_thread.kill
- end
-
- private
-
- def run
- spin_worker_threads
-
- loop do
- rd, = select([@signal_pipes[0], *@connections])
- if rd.delete @signal_pipes[0]
- break if read_signals == :shutdown
- end
- rd.each do |fd|
- begin
- if fd.handle.eof?
- remove_connection fd
- else
- read_connection fd
- end
- rescue Errno::ECONNRESET
- remove_connection fd
- end
- end
- end
- join_worker_threads(@shutdown_timeout)
- ensure
- @shutdown_queue.push :shutdown
- end
-
- def read_connection(fd)
- @buffers[fd] << fd.read(DEFAULT_BUFFER)
- while(frame = slice_frame!(@buffers[fd]))
- @logger.debug "#{self} is processing a frame"
- @worker_queue.push [:frame, fd, frame]
- end
- end
-
- def spin_worker_threads
- @logger.debug "#{self} is spinning up worker threads"
- @worker_threads = []
- @num_threads.times do
- @worker_threads << spin_thread
- end
- end
-
- def spin_thread
- Worker.new(@processor, @transport_factory, @protocol_factory, @logger, @worker_queue).spawn
- end
-
- def signal(msg)
- @signal_queue << msg
- @signal_pipes[1].write " "
- end
-
- def read_signals
- # clear the signal pipe
- # note that since read_nonblock is broken in jruby,
- # we can only read up to a set number of signals at once
- sigstr = @signal_pipes[0].readpartial(1024)
- # now read the signals
- begin
- sigstr.length.times do
- signal, obj = @signal_queue.pop(true)
- case signal
- when :connection
- @connections << obj
- when :shutdown
- @shutdown_timeout = obj
- return :shutdown
- end
- end
- rescue ThreadError
- # out of signals
- # note that in a perfect world this would never happen, since we're
- # only reading the number of signals pushed on the pipe, but given the lack
- # of locks, in theory we could clear the pipe/queue while a new signal is being
- # placed on the pipe, at which point our next read_signals would hit this error
- end
- end
-
- def remove_connection(fd)
- # don't explicitly close it, a thread may still be writing to it
- @connections.delete fd
- @buffers.delete fd
- end
-
- def join_worker_threads(shutdown_timeout)
- start = Time.now
- @worker_threads.each do |t|
- if shutdown_timeout > 0
- timeout = (start + shutdown_timeout) - Time.now
- break if timeout <= 0
- t.join(timeout)
- else
- t.join
- end
- end
- kill_worker_threads
- end
-
- def kill_worker_threads
- @worker_threads.each do |t|
- t.kill if t.status
- end
- @worker_threads.clear
- end
-
- def slice_frame!(buf)
- if buf.length >= 4
- size = buf.unpack('N').first
- if buf.length >= size + 4
- buf.slice!(0, size + 4)
- else
- nil
- end
- else
- nil
- end
- end
-
- class Worker # :nodoc:
- def initialize(processor, transport_factory, protocol_factory, logger, queue)
- @processor = processor
- @transport_factory = transport_factory
- @protocol_factory = protocol_factory
- @logger = logger
- @queue = queue
- end
-
- def spawn
- Thread.new do
- @logger.debug "#{self} is spawning"
- run
- end
- end
-
- private
-
- def run
- loop do
- cmd, *args = @queue.pop
- case cmd
- when :shutdown
- @logger.debug "#{self} is shutting down, goodbye"
- break
- when :frame
- fd, frame = args
- begin
- otrans = @transport_factory.get_transport(fd)
- oprot = @protocol_factory.get_protocol(otrans)
- membuf = MemoryBufferTransport.new(frame)
- itrans = @transport_factory.get_transport(membuf)
- iprot = @protocol_factory.get_protocol(itrans)
- @processor.process(iprot, oprot)
- rescue => e
- @logger.error "#{Thread.current.inspect} raised error: #{e.inspect}\n#{e.backtrace.join("\n")}"
- end
- end
- end
- end
- end
- end
- end
-end
projects / ceph.git / blobdiff