# -*- encoding: binary -*- # Copyright (C) 2013, Eric Wong and all contributors # License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt) class Yahns::Queue < SleepyPenguin::Epoll::IO # :nodoc: include SleepyPenguin attr_accessor :fdmap # Yahns::Fdmap # public QEV_QUIT = Epoll::OUT # Level Trigger for QueueQuitter QEV_RD = Epoll::IN | Epoll::ONESHOT QEV_WR = Epoll::OUT | Epoll::ONESHOT QEV_RDWR = QEV_RD | QEV_WR def self.new super(SleepyPenguin::Epoll::CLOEXEC) end # for HTTP and HTTPS servers, we rely on the io writing to us, first # flags: QEV_RD/QEV_WR (usually QEV_RD) def queue_add(io, flags) @fdmap.add(io) epoll_ctl(Epoll::CTL_ADD, io, flags) end def thr_init Thread.current[:yahns_rbuf] = "" Thread.current[:yahns_fdmap] = @fdmap end # use only before hijacking, once hijacked, io may be unusable to us def queue_del(io) @fdmap.forget(io) epoll_ctl(Epoll::CTL_DEL, io, 0) end # returns an array of infinitely running threads def worker_thread(logger, max_events) Thread.new do thr_init begin epoll_wait(max_events) do |_, io| # don't care for flags for now case rv = io.yahns_step when :wait_readable epoll_ctl(Epoll::CTL_MOD, io, QEV_RD) when :wait_writable epoll_ctl(Epoll::CTL_MOD, io, QEV_WR) when :wait_readwrite epoll_ctl(Epoll::CTL_MOD, io, QEV_RDWR) when :ignore # only used by rack.hijack # we cannot call Epoll::CTL_DEL after hijacking, the hijacker # may have already closed it Likewise, io.fileno is not # expected to work, so we had to erase it from fdmap before hijack when nil, :close # this must be the ONLY place where we call IO#close on # things that got inside the queue @fdmap.sync_close(io) else raise "BUG: #{io.inspect}#yahns_step returned: #{rv.inspect}" end end rescue => e Yahns::Log.exception(logger, 'queue loop', e) end while true end end end