diff options
author | Matthew Draper <matthew@trebex.net> | 2016-10-01 00:57:08 +0930 |
---|---|---|
committer | GitHub <noreply@github.com> | 2016-10-01 00:57:08 +0930 |
commit | 8356dcac9b2e8239ade7ead15dee6ab08be605b7 (patch) | |
tree | ddeb946045083dff14c4d5cdc600b291be4e75cd /actioncable/lib/action_cable/connection/stream_event_loop.rb | |
parent | 4f2d1d6c4c126783129a5e2283a5290a8d392505 (diff) | |
parent | 5d92089bcae42275953ad310b24f5dd66ff0ec6e (diff) | |
download | rails-8356dcac9b2e8239ade7ead15dee6ab08be605b7.tar.gz rails-8356dcac9b2e8239ade7ead15dee6ab08be605b7.tar.bz2 rails-8356dcac9b2e8239ade7ead15dee6ab08be605b7.zip |
Merge pull request #26646 from matthewd/cable-buffer
Buffer writes to the cable sockets
Diffstat (limited to 'actioncable/lib/action_cable/connection/stream_event_loop.rb')
-rw-r--r-- | actioncable/lib/action_cable/connection/stream_event_loop.rb | 46 |
1 files changed, 38 insertions, 8 deletions
diff --git a/actioncable/lib/action_cable/connection/stream_event_loop.rb b/actioncable/lib/action_cable/connection/stream_event_loop.rb index 106b948c45..eec24638b6 100644 --- a/actioncable/lib/action_cable/connection/stream_event_loop.rb +++ b/actioncable/lib/action_cable/connection/stream_event_loop.rb @@ -5,7 +5,7 @@ module ActionCable module Connection class StreamEventLoop def initialize - @nio = @thread = nil + @nio = @executor = @thread = nil @map = {} @stopping = false @todo = Queue.new @@ -20,13 +20,14 @@ module ActionCable def post(task = nil, &block) task ||= block - Concurrent.global_io_executor << task + spawn + @executor << task end def attach(io, stream) @todo << lambda do - @map[io] = stream - @nio.register(io, :r) + @map[io] = @nio.register(io, :r) + @map[io].value = stream end wakeup end @@ -39,6 +40,15 @@ module ActionCable wakeup end + def writes_pending(io) + @todo << lambda do + if monitor = @map[io] + monitor.interests = :rw + end + end + wakeup + end + def stop @stopping = true wakeup if @nio @@ -52,6 +62,13 @@ module ActionCable return if @thread && @thread.status @nio ||= NIO::Selector.new + + @executor ||= Concurrent::ThreadPoolExecutor.new( + min_threads: 1, + max_threads: 10, + max_queue: 0, + ) + @thread = Thread.new { run } return true @@ -77,12 +94,25 @@ module ActionCable monitors.each do |monitor| io = monitor.io - stream = @map[io] + stream = monitor.value begin - stream.receive io.read_nonblock(4096) - rescue IO::WaitReadable - next + if monitor.writable? + if stream.flush_write_buffer + monitor.interests = :r + end + next unless monitor.readable? + end + + incoming = io.read_nonblock(4096, exception: false) + case incoming + when :wait_readable + next + when nil + stream.close + else + stream.receive incoming + end rescue # We expect one of EOFError or Errno::ECONNRESET in # normal operation (when the client goes away). But if |