Introduction

Now and then, I like to go on a performance improvement hunt because life isn't just about adding new features. Recently, I have been focusing on enhancing efficiency, particularly regarding CPU and memory usage in Karafka. Three of my recent pull requests (PR 117, PR 118, PR 123), have made some minor improvements, and this article is about them. These changes help reduce memory allocation and improve time tracking management in Karafka and WaterDrop.

Most of the time, such improvements are not significant, but when applied in crucial places, they can make things a bit faster.

When doing OSS, I think of myself as a middleman. Karafka runs in tens of thousands of processes, and improvements affecting the consumption or production of messages (especially when applicable per message) can make a difference when multiplied.

Shaving Memory Allocations

PR 117 targets memory savings by optimizing instrumentation data handling. The primary change involves reducing unnecessary array allocations during instrumentation. Previously, every instrumentation event would allocate a new array, leading to excessive memory usage. The updated code minimizes these allocations by inlining the tracking code.

Here's a simple instrumentation layer one could build:

# This is a simplified bus without too many features
#
# It supports simple instrumentation piping details to listeners and returning the results
class Notifications
  def initialize(*listeners)
    @listeners = listeners
  end

  def instrument(name, &block)
    result, time = measure_time_taken(&block)

    details = { name: name, result: result, time: time }

    @listeners.each do |listener|
      listener.call(details)
    end

    result
  end

  private

  def monotonic_now
    ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) * 1_000
  end

  def measure_time_taken
    start = monotonic_now
    result = yield
    [result, monotonic_now - start]
  end
end

It's a fairly simple instrumentation that allows you to have listeners and wrap your code with it for tracking:

bus = Notifications.new(
  ->(details) { puts "Event: #{details[:name]} occurred" },
  ->(details) { puts("  and it took #{details[:time]}ms to run") }
)

bus.instrument('my.random.sleep') do
  sleep_time = rand
  sleep(sleep_time)
  sleep_time
end

# Event: my.random.sleep occurred
#   and it took 799.0296444892883ms
# => 0.7981784182914173

It works and is actually quite fast:

bus = Notifications.new
EVENT = 'test'
puts Benchmark.measure { 1_000_000.times { bus.instrument(EVENT) {} } }

# 0.538294   0.000000   0.538294 (  0.539663)

However, there is one thing that is fundamentally wrong with this code, and that is time tracking. It may not be visible at first, but when you start counting object allocations, this is what you end up with:

GC.disable

def count_objects_created
  # Take a snapshot of the current object counts
  initial_counts = ObjectSpace.each_object.count

  yield if block_given?

  # Take another snapshot after running the block
  final_counts = ObjectSpace.each_object.count

  # Calculate the difference
  new_objects_created = final_counts - initial_counts

  puts "Number of objects created: #{new_objects_created}"
end

bus = Notifications.new
EVENT = 'test'

count_objects_created do
  1_000_000.times { bus.instrument(EVENT){} }
end

# Number of objects created: 2000002

there are twice as many objects and all of this because of the return value of the time measurement:

def measure_time_taken
  start = monotonic_now
  result = yield
  [result, monotonic_now - start]
end

It returns an array of results and time measurements. It may not be a lot but nonetheless, lets inline this instead of delegating to a method:

def instrument(name)
  start = monotonic_now
  result = yield
  time = monotonic_now - start

  details = { name: name, result: result, time: time }

  @listeners.each do |listener|
    listener.call(details)
  end

  result
end

With this change we no longer allocate the arrays:

bus = Notifications.new
EVENT = 'test'

count_objects_created do
  1_000_000.times { bus.instrument(EVENT){} }
end

# Number of objects created: 1000002

*Numer of objects allocated per one million of instrumentation calls.

You may question why it is relevant and whether it provides significant benefits. I would say that it depends. Karafka is heavily instrumented, and under heavy load, this simple change saves 20-30 thousand allocations per second of execution.

If a Tree Falls: Fast Paths for Unmonitored Events

As I mentioned, Karafka and WaterDrop are heavily instrumented. Since different people can use different events for different use cases (logging, AppSignal instrumentation, or Web UI), there is no silver bullet regarding what to instrument and what not to do. This means that Karafka emits many events during its execution. Same with WaterDrop. During my optimization session, I wondered if there's even a point in measuring and publishing the instrumentation results when no one listens. And this is what the PR 123 is about. If no one is listening, there is no point in making any sound.

Below you can find a simplified previous version of the instrument method. Similar code can be found for example, in dry-monitor and dry-events

# Before
def instrument(name)
  start = monotonic_now
  result = yield
  time = monotonic_now - start

  details = { name: name, result: result, time: time }

  @listeners[name].each do |listener|
    listener.call(details)
  end

  result
end

It's nice and fairly fast. But in the case of publishing many events, it may be optimized as long as we have a way to check if there are any listeners:

# After
def instrument(name)
  listeners = @listeners[name]

  # Measure nothing since no one will use it
  return yield unless listeners

  start = monotonic_now
  result = yield
  time = monotonic_now - start

  details = { name: name, result: result, time: time }

  @listeners[name].each do |listener|
    listener.call(details)
  end

  result
end

For a fast track instrumentation cases, the changed code is over 2.5x faster:

# Based on the PR change measurements

1.830753   0.005683   1.836436 (  1.843820)
# vs.
0.759046   0.000000   0.759046 (  0.759051)

*Time difference when executing instrumentation in the fast-path scenario 1 million times.

This also saves on memory allocations because when no one will use the results, the event details are not being built at all.

Timing Time

The last optimization in this batch was about time measurements. Karafka uses two types of time tracking:

• A monotonic clock is used to measure distance in time in a way that would not break because of time zone changes, yearly second alignments, or anything else. The CPU monotonic clock is a clock source that provides a monotonically increasing time value, which means it will never go backward.

• Regular UTC-based time: A normal clock is used to take snapshots when something has happened.

def monotonic_now
  # Returns float with seconds, we wanted in milliseconds
  ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) * 1_000
end

def float_now
  ::Time.now.utc.to_f
end

The above methods are fairly simple but they do contain space for two improvements:

The first thing that can be done is the elimination of the millisecond multiplication. Ruby now supports the float_millisecond time that allows us not to deal with the multiplication explicitly:

def monotonic_now
  ::Process.clock_gettime(::Process::CLOCK_MONOTONIC, :float_millisecond)
end

The difference is not significant (if any), but this at least simplifies the code and makes it more explicit.

While the above did not yield performance gains, the latter is much more interesting:

def float_now
  ::Time.now.utc.to_f
end

The above code creates a few objects down the road and then casts the time into a float. This may not seem expensive at first, but keep in mind that this code may run thousands of times per second in Karafka. Since our goal is to get a float time, this can also be replaced with a system clock invocation:

def float_now
  ::Process.clock_gettime(Process::CLOCK_REALTIME)
end

The difference in terms of performance is big:

6.405367   0.546916   6.952283 (  6.952844)
# vs.
1.887427   0.003451   1.890878 (  1.897118)

*Time difference when fetching the real-time in both ways.

Using the process clock is 3,4x times faster!

Conclusion

The recent optimizations in Karafka, shown in the PRs mentioned, reflect my commitment to pushing this project forward. As I refine Karafka, I aim to deliver top-notch data-streaming tools for the Ruby community. Every optimization brings Karafka Core closer to achieving the best possible performance and efficiency.