ActionCable Streaming Benchmark

ac_streams_bench.rb

# rubocop disable:all
begin
  require 'bundler/inline'
rescue LoadError => e
  $stderr.puts 'Bundler version 1.10 or later is required. Please update your Bundler'
  raise e
end

gemfile(true) do
  source 'https://rubygems.org'
  gem 'actioncable', '~>5.0'
  gem 'benchmark-ips'
end

require 'json'
require 'action_cable'
require 'logger'
require 'benchmark/ips'
require 'benchmark'

class Measure
  def self.run(gc: :enable)
    if gc == :disable
      GC.disable
    elsif gc == :enable
      GC.start
    end

    memory_before = `ps -o rss= -p #{Process.pid}`.to_i / 1024
    gc_stat_before = GC.stat
    time = Benchmark.realtime do
      yield
    end

    gc_stat_after = GC.stat

    GC.start if gc == :enable

    memory_after = `ps -o rss= -p #{Process.pid}`.to_i / 1024

    $stdout.puts(
      {
        gc: gc,
        time: time.round(2),
        gc_major_count: gc_stat_after[:major_gc_count].to_i - gc_stat_before[:major_gc_count].to_i,
        gc_minor_count: gc_stat_after[:minor_gc_count].to_i - gc_stat_before[:minor_gc_count].to_i,
        memory: "%d MB" % (memory_after - memory_before)
      }.to_json
    )
  end
end

class FakeSocket
  def transmit(msg)
    msg.to_s
  end
end

class FakeConnection < ActionCable::Connection::Base
  def initialize
    @coder = ActiveSupport::JSON
  end

  def websocket
    @websocket ||= FakeSocket.new
  end

  def worker_pool
    @worker_pool ||= ActionCable::Server::Worker.new(max_size: 4)
  end

  def logger
    @logger ||= Logger.new(nil)
  end
end

class SimpleSubscriberMap < ActionCable::SubscriptionAdapter::SubscriberMap
  def invoke_callback(callback, message)
    callback[:connection].worker_pool.async_invoke(
      self,
      :transmit,
      message,
      connection: callback[:connection],
    )
  end

  def transmit(message, connection:)
    connection.send(:websocket).transmit(
      "{\"identifier\": #{callback[:id]},\"message\": #{message}}"
    )
  end
end

class DecodingSubscriberMap < SimpleSubscriberMap
  def transmit(message, connection:)
    connection.transmit(
      identifier: callback[:id],
      message: ActiveSupport::JSON.decode(message)
    )
  end
end

class FakeChannel < ActionCable::Channel::Base
  def delegate_connection_identifiers
    # noop
  end

  def subscribe_to_channel
    # noop
  end
end

decoding_map = DecodingSubscriberMap.new
simple_map = SimpleSubscriberMap.new
base_map = ActionCable::SubscriptionAdapter::SubscriberMap.new

message = { text: "ActionCable should be better" }.to_json

N = 100

Benchmark.ips do |x|
  # Very simple transmitter: just broadcast message as is
  x.report('SimpleSubscriberMap') do
    connection = FakeConnection.new
    simple_handler = { connection: connection, id: 'FakeChannel' }

    N.times do
      simple_map.invoke_callback(simple_handler, message)
    end

    connection.worker_pool.executor.shutdown
    connection.worker_pool.executor.wait_for_termination
  end

  # Simple transmitter with JSON round-trip
  x.report('DecodingSubscriberMap') do
    connection = FakeConnection.new
    simple_handler = { connection: connection, id: 'FakeChannel' }

    N.times do
      decoding_map.invoke_callback(simple_handler, message)
    end
  
    connection.worker_pool.executor.shutdown
    connection.worker_pool.executor.wait_for_termination    
  end

  # Current ActionCable callback hell implementation
  x.report('SubscriberMap') do
    connection = FakeConnection.new
    channel = FakeChannel.new(connection, 'FakeChannel')

    callback_handler = channel.send(:worker_pool_stream_handler, 'test', nil)

    N.times do
      base_map.invoke_callback(callback_handler, message)
    end

    connection.worker_pool.executor.shutdown
    connection.worker_pool.executor.wait_for_termination
  end
end