mitigate ffi bugs (#491)

Author: mensfeld

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 0.18.0 (Unreleased)
 - [Enhancement] Update `librdkafka` to `2.5.0`
+- [Fix] Mitigate a case where FFI would not restart the background events callback dispatcher in forks
 
 ## 0.17.0 (2024-08-03)
 - [Feature] Add `#seek_by` to be able to seek for a message by topic, partition and offset (zinahia)

lib/rdkafka/callbacks.rb

@@ -2,7 +2,6 @@
 
 module Rdkafka
   module Callbacks
-
     # Extracts attributes of a rd_kafka_topic_result_t
     #
     # @private
@@ -149,13 +148,6 @@ def initialize(event_ptr)
       end
     end
 
-    # FFI Function used for Create Topic and Delete Topic callbacks
-    BackgroundEventCallbackFunction = FFI::Function.new(
-        :void, [:pointer, :pointer, :pointer]
-    ) do |client_ptr, event_ptr, opaque_ptr|
-      BackgroundEventCallback.call(client_ptr, event_ptr, opaque_ptr)
-    end
-
     # @private
     class BackgroundEventCallback
       def self.call(_, event_ptr, _)
@@ -348,13 +340,6 @@ def self.process_describe_acl(event_ptr)
       end
     end
 
-    # FFI Function used for Message Delivery callbacks
-    DeliveryCallbackFunction = FFI::Function.new(
-        :void, [:pointer, :pointer, :pointer]
-    ) do |client_ptr, message_ptr, opaque_ptr|
-      DeliveryCallback.call(client_ptr, message_ptr, opaque_ptr)
-    end
-
     # @private
     class DeliveryCallback
       def self.call(_, message_ptr, opaque_ptr)
@@ -387,5 +372,44 @@ def self.call(_, message_ptr, opaque_ptr)
         end
       end
     end
+
+    @@mutex = Mutex.new
+    @@current_pid = nil
+
+    class << self
+      # Defines or recreates after fork callbacks that require FFI thread so the callback thread
+      # is always correctly initialized
+      #
+      # @see https://github.com/ffi/ffi/issues/1114
+      def ensure_ffi_running
+        @@mutex.synchronize do
+          return if @@current_pid == ::Process.pid
+
+          if const_defined?(:BackgroundEventCallbackFunction, false)
+            send(:remove_const, :BackgroundEventCallbackFunction)
+            send(:remove_const, :DeliveryCallbackFunction)
+          end
+
+          # FFI Function used for Create Topic and Delete Topic callbacks
+          background_event_callback_function = FFI::Function.new(
+              :void, [:pointer, :pointer, :pointer]
+          ) do |client_ptr, event_ptr, opaque_ptr|
+            BackgroundEventCallback.call(client_ptr, event_ptr, opaque_ptr)
+          end
+
+          # FFI Function used for Message Delivery callbacks
+          delivery_callback_function = FFI::Function.new(
+              :void, [:pointer, :pointer, :pointer]
+          ) do |client_ptr, message_ptr, opaque_ptr|
+            DeliveryCallback.call(client_ptr, message_ptr, opaque_ptr)
+          end
+
+          const_set(:BackgroundEventCallbackFunction, background_event_callback_function)
+          const_set(:DeliveryCallbackFunction, delivery_callback_function)
+
+          @@current_pid = ::Process.pid
+        end
+      end
+    end
   end
 end

lib/rdkafka/config.rb

@@ -146,6 +146,8 @@ def self.opaques
     #
     # @return [Config]
     def initialize(config_hash = {})
+      Callbacks.ensure_ffi_running
+
       @config_hash = DEFAULT_CONFIG.merge(config_hash)
       @consumer_rebalance_listener = nil
       @consumer_poll_set = true

spec/rdkafka/admin_spec.rb

@@ -737,4 +737,18 @@
       end
     end
   end
+
+  context "when operating from a fork" do
+    # @see https://github.com/ffi/ffi/issues/1114
+    it 'expect to be able to create topics and run other admin operations without hanging' do
+      # If the FFI issue is not mitigated, this will hang forever
+      pid = fork do
+        admin
+          .create_topic(topic_name, topic_partition_count, topic_replication_factor)
+          .wait
+      end
+
+      Process.wait(pid)
+    end
+  end
 end