Tag: bug

Ruby concurrency is hard: how I became a Ruby on Rails contributor

For the past several weeks, I've been trying to fix a cranky spec in Karafka integrations suite, which in the end, lead me to become a Ruby on Rails micro-contributor and submitting similar fix to several other high-popularity projects from the Ruby ecosystem. Here's my story of trying to make sense of my specs and Ruby concurrency.

Ephemeral bug from a test-suite

Karafka is a Ruby and Rails multi-threaded efficient Kafka processing framework. To provide reliable OSS that is multi-threaded, I had to have the option to run my test suite concurrently to simulate how Karafka operates. Since it was a specific use case, I created my micro-framework.

Long story short: It runs end-to-end integration specs by running them in separate Ruby processes. Each starts Karafka, runs all the code in various configurations, connects to Kafka, checks assertions, and at the end, shuts down.

Such an approach allowed me to ensure that the process's whole lifecycle and its components work as expected. Specs are started with supervision, so in case of any hang, it will be killed after 5 minutes.

Karafka itself also has an internal shutdown supervisor. In case of a user shutdown request, if the shutdown takes longer than the defined expected time, Karafka will stop despite having things running. And this is what was happening with this single spec:

E, [2022-11-19T16:47:49.602718 #14843] ERROR -- : Forceful Karafka server stop
F, [2022-11-19T16:47:49.602825 #14843] FATAL -- : #<Karafka::Core::Monitoring::Event:0x0000562932d752b0 @id="error.occurred", @payload={:caller=>Karafka::Server, :error=>#<Karafka::Errors::ForcefulShutdownError: Karafka::Errors::ForcefulShutdownError>, :type=>"app.stopping.error"}>

This damn spec did not want to stop!

Many things are working under the hood:

  • workers that process jobs that could hang and force the process to wait
  • jobs queue that is also connected to the polling thread (to poll more data when no work is to be done)
  • listeners that poll data from Kafka that could hang
  • consumer groups with several threads polling Kafka data that might get stuck because of some underlying error
  • Other bugs in the coordination of work and states.

One thing that certainly worked was the process supervision that would forcefully kill it after 30 seconds.

Process shutdown coordination

The graceful shutdown of such a process takes work. When you have many connections to Kafka, upon a poorly organized shutdown, you may trigger several rebalances that may cause short-lived topics assignments causing nothing except friction and potentially blocking the whole process.

To mitigate this, Karafka shuts down actively and gracefully. That is, until the absolute end, it claims the ownership of given topics and partitions, actively waiting for all the current work to be finished. This looks more or less like so:

Note: Consumer groups internally in Karafka are a bit different than Kafka consumer groups. Here we focus on internal Karafka concepts.

Pinpointing the issue

After several failed attempts and fixing other bugs, I added a lot of extra instrumentation to check what Karafka hangs on. It was hanging because there were hanging listener threads!

As stated above, to close Karafka gracefully, all work from the jobs queue needs to be finished, and listeners that poll data from Kafka need to be able to exit the polling loops. It's all coordinated using a job queue. The job queue we're using is pretty complex with some blocking capabilities, and you can read about it here, but the interesting part of the code can be reduced to this:

@semaphores = Concurrent::Map.new { |h, k| h[k] = Queue.new }

Those queues are used as semaphores in the polling loops until all the current work is done. Since each Queue is assigned to a different subscription group within its thread and hidden behind a concurrent map, there should be no problem. Right?

Reproduction

Once I had my crazy suspicion, I decided to reduce it down to a proof of concept:

require 'concurrent-ruby'

100.times do
  ids = Set.new
  semaphores = Concurrent::Hash.new { |h, k| h[k] = Queue.new }

  100.times.map do
    Thread.new do
      ids << semaphores['test'].object_id
    end
  end.each(&:join)

  raise "I expected 1 semaphore but got #{ids.size}" if ids.size != 1
end

once executed, boom:

poc.rb:13:in `<main>': I expected 1 semaphore but got 2 (RuntimeError)

There is more than one semaphore for one listener! This caused Karafka to wait until forced to stop because the worker thread would use a different semaphore than the listener thread.

But how is that even possible?

Well, Concurrent::Hash and Concurrent::Map initialization is indeed thread-safe but not precisely as you would expect them to be. The docs state that:

This version locks against the object itself for every method call, ensuring only one thread can be reading or writing at a time. This includes iteration methods like #each, which takes the lock repeatedly when reading an item.

"only one thread can be reading or writing at a time". However, we are doing both at different times. Our code:

semaphores = Concurrent::Hash.new { |h, k| h[k] = Queue.new }

is actually equivalent to:

semaphores = Concurrent::Hash.new do |h, k|
  queue = Queue.new
  h[k] = queue
end

and the block content is not locked fully. One threads queue can overwrite the other if the Ruby scheduler stops the execution in the middle. Here's the flow of things happening in the form of a diagram:

Once in a while listener would receive a dangling queue object, effectively blocking the polling process.

Fixing the issue

This can be fixed either by replacing the Concurrent::Hash with Concurrent::Map and using the #compute_if_absent method or by introducing a lock inside of the Concurrent::Hash initialization block:

Concurrent::Map.new do |k, v|
  k.compute_if_absent(v) { [] }
end

mutex = Mutex.new

Concurrent::Hash.new do |k, v|
  mutex.synchronize do
    break k[v] if k.key?(v)
    k[v] = []
  end
end

Okay, but what does Ruby on Rails and other projects do with all of this?

Fixing the world

I figured out that if I made this mistake, maybe others did. I decided to check my local gems to find occurrences quickly. Inside my local gem cache, I executed the following code:

fgrep -R 'Concurrent::Hash.new {' ./
fgrep -R 'Concurrent::Hash.new do' ./
fgrep -R 'Concurrent::Map.new {' ./
fgrep -R 'Concurrent::Map.new do' ./

and validated that I'm not an isolated case. I wasn't alone!

Then using Sourcegraph I pinpointed a few projects that had the potential for fixes:

  • rails (activesupport and actionview)
  • i18n
  • dry-schema
  • finite_machine
  • graphql-ruby
  • rom-factory
  • apache whimsy
  • krane
  • puppet

I am not a domain expert in any of those, and understanding the severity of each was beyond my time constraints, but I decided to give it a shot.

Rails (ActiveSupport and ActionView)

Within Rails, this "pattern" was used twice: in ActiveSupport and ActionView.

In ActionView, it was used within a cache:

PREFIXED_PARTIAL_NAMES = Concurrent::Map.new do |h, k|
  h[k] = Concurrent::Map.new
end

and assuming that the cached result is stateless (same result each time for the same key), the issue could only cause an extra computation upon first parallel requests to this cache.

In the case of ActiveSupport, none of the concurrency code was needed, so I just replaced it with a simple Hash.

Both, luckily, were not that severe, though worth fixing nonetheless.

PR: https://github.com/rails/rails/pull/46536
PR: https://github.com/rails/rails/pull/46534

Both were merged, and this is how I became a Ruby on Rails contributor :)

i18n

This case was slightly more interesting because the concurrent cache stores all translations. In theory, this could cause similar leakage as in Karafka, effectively losing a language by loading it to a different Concurrent::Hash:

100.times.map do
  Thread.new do
    I18n.backend.store_translations(rand.to_s, :foo => { :bar => 'bar', :baz => 'baz' })
  end
end.each(&:join)

I18n.available_locales.count #=> 1

This could lead to hard-to-debug problems. Once in a while, your system could raise something like this:

:en is not a valid locale (I18n::InvalidLocale)

without an apparent reason, and this problem would go away after a restart.

PR: https://github.com/ruby-i18n/i18n/pull/644

dry-schema

Another cache case where the risk would revolve around double-computing.

PR: https://github.com/dry-rb/dry-schema/pull/440

rom-factory

This one is interesting! Let's reduce the code to a smaller POC first and see what will happen under heavy threading:

require 'singleton'
require 'concurrent-ruby'

class Sequences
  include Singleton

  attr_reader :registry

  def initialize
    reset
  end

  def next(key)
    registry[key] += 1
  end

  def reset
    @registry = Concurrent::Map.new { |h, k| h[k] = 0 }
    self
  end
end

seq = Sequences.instance

loop do
  100.times.map do
    Thread.new { seq.next('boom') }
  end.each(&:join)

  size = seq.registry['boom']

  raise "Wanted 100 but got #{size}" unless size == 100

  seq.reset
end
poc.rb:37:in `block in <main>': Wanted 100 but got 1 (RuntimeError)

The counter value gets biased. What is even more interesting is that making the map safe won't be enough:

@registry = Concurrent::Map.new { |h, k| h.compute_if_absent(k) { 0 } }
poc.rb:36:in `block in <main>': Wanted 100 but got 55 (RuntimeError)

there is one more "unsafe" method:

def next(key)
  registry[key] += 1
end

this operation also is not atomic, thus needs to be wrapped with a mutex:

def initialize
  @mutex = Mutex.new
  reset
end

def next(key)
  @mutex.synchronize do
    registry[key] += 1
  end
end

Only then is this code safe to be used.

https://github.com/rom-rb/rom-factory/pull/80

Other repositories

Summary

In my opinion, there are a few outcomes of this story:

  • Karafka has a solid test-suite!
  • If you are doing concurrency-related work, you better test it in a multi-threaded environment and test it well.
  • Concurrency is hard to many of us (maybe that's because we are special ;) ).
  • RTFM and read it well :)
  • Do not be afraid to help others by submitting pull requests!

On the other hand, looking at the frequency of this issue, it may be worth opening a discussion about changing this behavior and making the initialization fully locked.

Afterwords

Concurrent::Hash under cRuby is just a Hash. You can check it out here.


Cover photo by James Broad on Attribution-NonCommercial-ShareAlike 2.0 Generic (CC BY-NC-SA 2.0). Image has been cropped.

How requiring a gem can mess up your already running application

Introduction

Ruby's dynamic nature is both its advantage and disadvantage. Being able to reopen system classes during runtime, while useful, can also lead to unexpected behaviors. This article presents one such case: how just requiring a gem can mess things up in a completely different area of the application.

The bizzare error

Recently, after connecting the Diffend monitor into one of my systems, it started reporting a bizarre error:

uninitialized constant Whenever

whenever-1.0.0/lib/whenever/numeric.rb:3:in `respond_to?'
lib/ruby/2.7.0/bundler/settings.rb:368:in `=='
lib/ruby/2.7.0/bundler/settings.rb:368:in `=='
lib/ruby/2.7.0/bundler/settings.rb:368:in `converted_value'
lib/ruby/2.7.0/bundler/settings.rb:94:in `[]'
lib/ruby/2.7.0/bundler/fetcher.rb:80:in `'
lib/ruby/2.7.0/bundler/fetcher.rb:11:in `'
lib/ruby/2.7.0/bundler/fetcher.rb:9:in `'
diffend-monitor-0.2.36/lib/diffend/build_bundler_definition.rb:18:in `call'
diffend-monitor-0.2.36/lib/diffend/execute.rb:22:in `build_definition'
diffend-monitor-0.2.36/lib/diffend/execute.rb:12:in `call'
diffend-monitor-0.2.36/lib/diffend/track.rb:21:in `start'
diffend-monitor-0.2.36/lib/diffend/monitor.rb:42:in `block in '

the line in which it was happening was just a delegation to the Bundler API method:

::Bundler::Fetcher.disable_endpoint = nil

and in Bundler itself it is just an attr_accessor:

class << self
  attr_accessor :disable_endpoint, :api_timeout, :redirect_limit
end

So what does all of it has to do with the Whenever gem? Nothing.

We have nothing to do with Whenever but it does not mean Whenever has nothing to do with us.

Requiring a gem does not only mean that its code is being loaded. It also means that the gem can perform any operations it wants, whether legit or malicious.

When diffend-monitor is being required, it spins up its own Ruby thread and starts reporting data. And here is the moment when Whenever kicks in. It was being required after the monitor. Thus the monitor code was already running. In theory, those two should be separated entirely. Whenever and Diffend do entirely different things and they have their own namespaces.

It turns out, unfortunately, that Whenever is monkey patching Numeric class in an incorrect way:

Numeric.class_eval do
  def respond_to?(method, include_private = false)
    super || Whenever::NumericSeconds.public_method_defined?(method)
  end

  def method_missing(method, *args, &block)
    if Whenever::NumericSeconds.public_method_defined?(method)
      Whenever::NumericSeconds.new(self).send(method)
    else
      super
    end
  end
end

This patch seems to be safe, but there's a really big assumption made: Whenever::NumericSeconds needs to be accessible. If we look into the Whenever code loading file, we will notice, that the patch is required before Whenever::NumericSeconds comes to existence:

require 'whenever/numeric'
require 'whenever/numeric_seconds'

This means that any action that would invoke #method_missing after the first file is loaded, but before the second one, will fail.

Can it even happen? Absolutely! Ruby's require is not blocking. It means, that Ruby VM can stop the requiring after any of the files and switch context to do other things in other threads.

When the above is understood, building a reproduction code is just a matter of seconds:

Thread.new do
  while true
    begin
      1.respond_to?(:elo)
      sleep 0.00001
    rescue => e
      p e
    end
  end
end

sleep 0.2

require 'whenever'

Here's how it behaves when executed:

I’ve created an issue in Whenever, and hopefully, its maintainers will address it. Meanwhile there's one more question to ask: can we somehow address this problem, so it won't break our code?

Mitigating the issue before the library is patched

There is no silver bullet for this type of problem. As any gem can introduce their own patches to other classes, the potential problems are endless. In this particular case, the code ends up being “ok” once everything is loaded. What we’ve decided to do was pretty trivial. We’ve decided to give the app enough time to require all the things that could potentially break the execution:

Thread.new do
  sleep 0.5

  while true
    begin
      1.respond_to?(:elo)
      sleep 0.00001
    rescue => e
      p e
    end
  end
end

This sleep ensures that as long as nothing heavy happens during gems requirement via Bundler, we don't end up with partially loaded, broken monkey-patches while executing our own logic in a background thread.


Cover photo by Ruin Raider on Attribution-NonCommercial-NoDerivs 2.0 Generic (CC BY-NC-ND 2.0) license.

Copyright © 2022 Closer to Code

Theme by Anders NorenUp ↑