diff --git a/lib/ferrum/client.rb b/lib/ferrum/client.rb index 7b1341b3..d2f9d518 100644 --- a/lib/ferrum/client.rb +++ b/lib/ferrum/client.rb @@ -68,6 +68,7 @@ class Client def initialize(ws_url, options) @command_id = 0 + @command_id_mutex = Mutex.new @ws_url = ws_url @options = options @pendings = Concurrent::Hash.new @@ -154,8 +155,10 @@ def start end end + # Locked so two concurrent commands never share an id and read each + # other's responses from @pendings. def next_command_id - @command_id += 1 + @command_id_mutex.synchronize { @command_id += 1 } end def raise_browser_error(error) diff --git a/lib/ferrum/client/web_socket.rb b/lib/ferrum/client/web_socket.rb index 3f8a4de8..00afec41 100644 --- a/lib/ferrum/client/web_socket.rb +++ b/lib/ferrum/client/web_socket.rb @@ -30,7 +30,10 @@ def initialize(url, max_receive_size, logger) end max_receive_size ||= ::WebSocket::Driver::MAX_LENGTH - @driver = ::WebSocket::Driver.client(self, max_length: max_receive_size) + @driver = ::WebSocket::Driver.client(self, max_length: max_receive_size) + # websocket-driver holds no locks and is called from many threads: commands from + # callers, pong/close replies from the reader. One lock keeps frames from interleaving. + @driver_mutex = Mutex.new @messages = Queue.new @screenshot_commands = Concurrent::Hash.new if SKIP_LOGGING_SCREENSHOTS @@ -41,7 +44,7 @@ def initialize(url, max_receive_size, logger) start - @driver.start + @driver_mutex.synchronize { @driver.start } end def on_open(_event) @@ -76,7 +79,7 @@ def send_message(data) @screenshot_commands[data[:id]] = true if SKIP_LOGGING_SCREENSHOTS json = data.to_json - @driver.text(json) + @driver_mutex.synchronize { @driver.text(json) } @logger&.puts("\n\n▶ #{Utils::ElapsedTime.elapsed_time} #{json}") end @@ -87,7 +90,7 @@ def write(data) end def close - @driver.close + @driver_mutex.synchronize { @driver.close } end private @@ -98,7 +101,7 @@ def start data = @sock.readpartial(512) break unless data - @driver.parse(data) + @driver_mutex.synchronize { @driver.parse(data) } end rescue EOFError, Errno::ECONNRESET, Errno::EPIPE, IOError # rubocop:disable Lint/ShadowedException @messages.close diff --git a/spec/unit/client/web_socket_spec.rb b/spec/unit/client/web_socket_spec.rb new file mode 100644 index 00000000..db744ecb --- /dev/null +++ b/spec/unit/client/web_socket_spec.rb @@ -0,0 +1,51 @@ +# frozen_string_literal: true + +describe Ferrum::Client::WebSocket do + # `allocate` skips the constructor so no socket connects; the ivars the send + # path relies on are set by hand. + subject(:web_socket) { described_class.allocate } + + # Stands in for the websocket driver and measures how many threads are inside + # `text` at once — anything above 1 corrupts the frame stream on a real driver. + let(:frame_probe_class) do + Class.new do + def initialize + @frames = Thread::Queue.new + @active = Concurrent::AtomicFixnum.new(0) + @peak = Concurrent::AtomicFixnum.new(0) + end + + def text(json) + level = @active.increment + @peak.update { |peak| [peak, level].max } + sleep 0.005 + @frames << json + @active.decrement + end + + def frames = Array.new(@frames.size) { @frames.pop } + + def peak_concurrent_writes = @peak.value + end + end + + let(:frame_probe) { frame_probe_class.new } + + before do + web_socket.instance_variable_set(:@driver, frame_probe) + web_socket.instance_variable_set(:@driver_mutex, Mutex.new) + web_socket.instance_variable_set(:@screenshot_commands, Concurrent::Hash.new) + end + + describe "#send_message" do + it "hands the driver the message as JSON" do + web_socket.send_message(id: 1) + expect(frame_probe.frames).to eq(['{"id":1}']) + end + + it "writes one frame at a time" do + Array.new(8) { |i| Thread.new { web_socket.send_message(id: i) } }.each(&:join) + expect(frame_probe.peak_concurrent_writes).to eq(1) + end + end +end diff --git a/spec/unit/client_spec.rb b/spec/unit/client_spec.rb new file mode 100644 index 00000000..29f800f9 --- /dev/null +++ b/spec/unit/client_spec.rb @@ -0,0 +1,51 @@ +# frozen_string_literal: true + +describe Ferrum::Client do + # `allocate` skips the constructor so no browser boots; the ivars the id + # mint relies on are set by hand. + subject(:client) { described_class.allocate } + + before do + client.instance_variable_set(:@command_id, 0) + client.instance_variable_set(:@command_id_mutex, Mutex.new) + end + + describe "#build_message" do + it "carries the method and params" do + expect(client.build_message("Page.enable", foo: 1)).to include(method: "Page.enable", params: { foo: 1 }) + end + + it "mints sequential ids for sequential commands" do + ids = Array.new(3) { client.build_message("Page.enable", {})[:id] } + expect(ids).to eq([1, 2, 3]) + end + + context "with the id increment window held open" do + # Stretches the read-modify-write of `@command_id += 1` so the + # lost-update race is deterministic instead of scheduler-dependent. + let(:racy_counter_class) do + Class.new do + attr_reader :to_i + + def initialize(value) + @to_i = value + end + + def +(other) + read = @to_i + sleep 0.005 + self.class.new(read + other) + end + end + end + + before { client.instance_variable_set(:@command_id, racy_counter_class.new(0)) } + + it "mints a unique id per command across concurrent threads" do + ids = Queue.new + Array.new(8) { Thread.new { ids << client.build_message("Page.enable", {})[:id].to_i } }.each(&:join) + expect(Array.new(8) { ids.pop }).to match_array((1..8).to_a) + end + end + end +end