Lock native only on closing and track usage (#264)

* Lock native only on closing and track usage

* fix desc

Author: mensfeld

lib/rdkafka/consumer.rb

@@ -27,9 +27,11 @@ def finalizer
     def close
       return if closed?
       ObjectSpace.undefine_finalizer(self)
-      @native_kafka.with_inner do |inner|
+
+      @native_kafka.synchronize do |inner|
         Rdkafka::Bindings.rd_kafka_consumer_close(inner)
       end
+
       @native_kafka.close
     end
 

lib/rdkafka/error.rb

@@ -92,4 +92,10 @@ def initialize(method)
       super("Illegal call to #{method.to_s} on a closed admin")
     end
   end
+
+  class ClosedInnerError < BaseError
+    def initialize
+      super("Illegal call to a closed inner librdkafka instance")
+    end
+  end
 end

lib/rdkafka/native_kafka.rb

@@ -10,6 +10,8 @@ def initialize(inner, run_polling_thread:)
       @access_mutex = Mutex.new
       # Lock around internal polling
       @poll_mutex = Mutex.new
+      # counter for operations in progress using inner
+      @operations_in_progress = 0
 
       if run_polling_thread
         # Start thread to poll client for delivery callbacks,
@@ -35,10 +37,26 @@ def initialize(inner, run_polling_thread:)
     end
 
     def with_inner
-      return if @inner.nil?
+      if @access_mutex.owned?
+        @operations_in_progress += 1
+      else
+        @access_mutex.synchronize { @operations_in_progress += 1 }
+      end
+
+      @inner.nil? ? raise(ClosedInnerError) : yield(@inner)
+    ensure
+      @operations_in_progress -= 1
+    end
 
+    def synchronize(&block)
       @access_mutex.synchronize do
-        yield @inner
+        # Wait for any commands using the inner to finish
+        # This can take a while on blocking operations like polling but is essential not to proceed
+        # with certain types of operations like resources destruction as it can cause the process
+        # to hang or crash
+        sleep(0.01) until @operations_in_progress.zero?
+
+        with_inner(&block)
       end
     end
 
@@ -53,31 +71,31 @@ def closed?
     def close(object_id=nil)
       return if closed?
 
-      @access_mutex.lock
+      synchronize do
+        # Indicate to the outside world that we are closing
+        @closing = true
 
-      # Indicate to the outside world that we are closing
-      @closing = true
+        if @polling_thread
+          # Indicate to polling thread that we're closing
+          @polling_thread[:closing] = true
 
-      if @polling_thread
-        # Indicate to polling thread that we're closing
-        @polling_thread[:closing] = true
-
-        # Wait for the polling thread to finish up,
-        # this can be aborted in practice if this
-        # code runs from a finalizer.
-        @polling_thread.join
-      end
+          # Wait for the polling thread to finish up,
+          # this can be aborted in practice if this
+          # code runs from a finalizer.
+          @polling_thread.join
+        end
 
-      # Destroy the client after locking both mutexes
-      @poll_mutex.lock
+        # Destroy the client after locking both mutexes
+        @poll_mutex.lock
 
-      # This check prevents a race condition, where we would enter the close in two threads
-      # and after unlocking the primary one that hold the lock but finished, ours would be unlocked
-      # and would continue to run, trying to destroy inner twice
-      return unless @inner
+        # This check prevents a race condition, where we would enter the close in two threads
+        # and after unlocking the primary one that hold the lock but finished, ours would be unlocked
+        # and would continue to run, trying to destroy inner twice
+        return unless @inner
 
-      Rdkafka::Bindings.rd_kafka_destroy(@inner)
-      @inner = nil
+        Rdkafka::Bindings.rd_kafka_destroy(@inner)
+        @inner = nil
+      end
     end
   end
 end

spec/rdkafka/consumer_spec.rb

@@ -286,6 +286,29 @@ def send_one_message(val)
         consumer.poll(100)
       }.to raise_error(Rdkafka::ClosedConsumerError, /poll/)
     end
+
+    context 'when there are outgoing operations in other threads' do
+      it 'should wait and not crash' do
+        times = []
+
+        # Run a long running poll
+        thread = Thread.new do
+          times << Time.now
+          consumer.subscribe("empty_test_topic")
+          times << Time.now
+          consumer.poll(1_000)
+          times << Time.now
+        end
+
+        # Make sure it starts before we close
+        sleep(0.1)
+        consumer.close
+        close_time = Time.now
+        thread.join
+
+        times.each { |op_time| expect(op_time).to be < close_time }
+      end
+    end
   end
 
   describe "#commit, #committed and #store_offset" do