Try queue

Author: comandeo-mongo

lib/mongo/cluster/reapers/cursor_reaper.rb

@@ -34,11 +34,6 @@ class CursorReaper
       # @since 2.3.0
       FREQUENCY = 1.freeze
 
-      # Symbol for separating parts of scheduled kill spec messages.
-      #
-      # @api private
-      MSG_SEPARATOR = "\t"
-
       # Create a cursor reaper.
       #
       # @param [ Cluster ] cluster The cluster.
@@ -49,27 +44,18 @@ def initialize(cluster)
         @to_kill = {}
         @active_cursor_ids = Set.new
         @mutex = Mutex.new
-        @pipe_read, @pipe_write = IO.pipe
+        @kill_spec_queue = Queue.new
       end
 
       attr_reader :cluster
 
       # Schedule a kill cursors operation to be eventually executed.
       #
       # @param [ Cursor::KillSpec ] kill_spec The kill specification.
-      # @param [ Mongo::Server ] server The server to send the kill cursors
-      #   operation to.
       #
       # @api private
-      def schedule_kill_cursor(kill_spec, server)
-        msg = [
-          server.address.seed,
-          kill_spec.cursor_id,
-          kill_spec.coll_name,
-          kill_spec.db_name,
-          kill_spec.service_id
-        ].join(MSG_SEPARATOR)
-        @pipe_write.puts(msg)
+      def schedule_kill_cursor(kill_spec)
+        @kill_spec_queue << kill_spec
       end
 
       # Register a cursor id as active.
@@ -126,20 +112,11 @@ def unregister_cursor(id)
       #
       # @api private
       def read_scheduled_kill_specs
-        while select_resp = IO.select([@pipe_read], nil, nil, 0.1)
-          readable_io = select_resp&.first&.first
-          next if readable_io.nil?
-          msg = readable_io.gets.strip
-          seed, cursor_id, coll_name, db_name, service_id = msg.split(MSG_SEPARATOR)
-          kill_spec = Cursor::KillSpec.new(
-            cursor_id: cursor_id.to_i,
-            coll_name: coll_name,
-            db_name: db_name,
-            service_id: service_id
-          )
+        while !@kill_spec_queue.empty?
+          kill_spec = @kill_spec_queue.pop
           if @active_cursor_ids.include?(kill_spec.cursor_id)
-            @to_kill[seed] ||= Set.new
-            @to_kill[seed] << kill_spec
+            @to_kill[kill_spec.server_seed] ||= Set.new
+            @to_kill[kill_spec.server_seed] << kill_spec
           end
         end
       end

lib/mongo/cursor.rb

@@ -84,9 +84,8 @@ def initialize(view, result, server, options = {})
       @session = @options[:session]
       unless closed?
         register
-        ObjectSpace.define_finalizer(self, self.class.finalize(kill_spec,
+        ObjectSpace.define_finalizer(self, self.class.finalize(kill_spec(server),
           cluster,
-          server,
           @session))
       end
     end
@@ -107,12 +106,12 @@ def initialize(view, result, server, options = {})
     # @return [ Proc ] The Finalizer.
     #
     # @api private
-    def self.finalize(kill_spec, cluster, server, session)
+    def self.finalize(kill_spec, cluster, session)
       unless KillSpec === kill_spec
         raise ArgumentError, "First argument must be a KillSpec: #{kill_spec.inspect}"
       end
       proc do
-        cluster.schedule_kill_cursor(kill_spec, server)
+        cluster.schedule_kill_cursor(kill_spec)
         session.end_session if session && session.implicit?
       end
     end
@@ -367,12 +366,13 @@ def get_more
     end
 
     # @api private
-    def kill_spec
+    def kill_spec(server)
       KillSpec.new(
         cursor_id: id,
         coll_name: collection_name,
         db_name: database.name,
         service_id: initial_result.connection_description.service_id,
+        server_seed: server.address.seed,
       )
     end
 

lib/mongo/cursor/kill_spec.rb

@@ -25,25 +25,30 @@ class Cursor
     # @api private
     class KillSpec
 
-      def initialize(cursor_id:, coll_name:, db_name:, service_id:)
+      def initialize(cursor_id:, coll_name:, db_name:, service_id:, server_seed:)
         @cursor_id = cursor_id
         @coll_name = coll_name
         @db_name = db_name
         @service_id = service_id
+        @server_seed = server_seed
       end
 
-      attr_reader :cursor_id, :coll_name, :db_name, :service_id
+      attr_reader :cursor_id, :coll_name, :db_name, :service_id, :server_seed
 
       def ==(other)
-        cursor_id == other.cursor_id && coll_name == other.coll_name && db_name == other.db_name && service_id == other.service_id
+        cursor_id == other.cursor_id &&
+          coll_name == other.coll_name &&
+          db_name == other.db_name &&
+          service_id == other.service_id &&
+          server_seed == other.server_seed
       end
 
       def eql?(other)
         self.==(other)
       end
 
       def hash
-        [cursor_id, coll_name, db_name, service_id].compact.hash
+        [cursor_id, coll_name, db_name, service_id, server_seed].compact.hash
       end
     end
   end

spec/mongo/cluster/cursor_reaper_spec.rb

@@ -41,12 +41,12 @@
     let(:cursor_id) { 1 }
     let(:cursor_kill_spec_1) do
       Mongo::Cursor::KillSpec.new(
-        cursor_id: cursor_id, coll_name: 'c', db_name: 'd', service_id: nil,
+        cursor_id: cursor_id, coll_name: 'c', db_name: 'd', service_id: nil, server_seed: address.seed
       )
     end
     let(:cursor_kill_spec_2) do
       Mongo::Cursor::KillSpec.new(
-        cursor_id: cursor_id, coll_name: 'c', db_name: 'q', service_id: nil,
+        cursor_id: cursor_id, coll_name: 'c', db_name: 'q', service_id: nil, server_seed: address.seed
       )
     end
     let(:to_kill) { reaper.instance_variable_get(:@to_kill)}
@@ -60,7 +60,7 @@
       context 'when there is not a list already for the server' do
 
         before do
-          reaper.schedule_kill_cursor(cursor_kill_spec_1, server)
+          reaper.schedule_kill_cursor(cursor_kill_spec_1)
           reaper.read_scheduled_kill_specs
         end
 
@@ -73,9 +73,9 @@
       context 'when there is a list of ops already for the server' do
 
         before do
-          reaper.schedule_kill_cursor(cursor_kill_spec_1, server)
+          reaper.schedule_kill_cursor(cursor_kill_spec_1)
           reaper.read_scheduled_kill_specs
-          reaper.schedule_kill_cursor(cursor_kill_spec_2, server)
+          reaper.schedule_kill_cursor(cursor_kill_spec_2)
           reaper.read_scheduled_kill_specs
         end
 
@@ -87,7 +87,7 @@
         context 'when the same op is added more than once' do
 
           before do
-            reaper.schedule_kill_cursor(cursor_kill_spec_2, server)
+            reaper.schedule_kill_cursor(cursor_kill_spec_2)
             reaper.read_scheduled_kill_specs
           end
 
@@ -102,7 +102,7 @@
     context 'when the cursor is not on the list of active cursors' do
 
       before do
-        reaper.schedule_kill_cursor(cursor_kill_spec_1, server)
+        reaper.schedule_kill_cursor(cursor_kill_spec_1)
       end
 
       it 'does not add the kill cursors op spec to the list' do
@@ -193,8 +193,11 @@
     around do |example|
       authorized_collection.insert_many(docs)
       periodic_executor.stop!
-      cluster.schedule_kill_cursor(cursor.kill_spec,
-                                   cursor.instance_variable_get(:@server))
+      cluster.schedule_kill_cursor(
+        cursor.kill_spec(
+          cursor.instance_variable_get(:@server)
+        )
+      )
       periodic_executor.flush
       example.run
       periodic_executor.run!

spec/mongo/cursor_spec.rb

@@ -331,8 +331,9 @@
 
       before do
         authorized_collection.insert_many(documents)
-        cluster.schedule_kill_cursor(cursor.kill_spec,
-                                     cursor.instance_variable_get(:@server))
+        cluster.schedule_kill_cursor(
+          cursor.kill_spec(cursor.instance_variable_get(:@server))
+        )
       end
 
       let(:view) do