Add the Scalient::Concurrent::CollectionRetrofit#cancel method so that takers can receive cancellation errors from adders

Author: carsomyr

app/concerns/scalient/concurrent/collection_retrofit.rb

@@ -24,6 +24,8 @@ module CollectionRetrofit
         if !include?(::MonitorMixin)
           send(:include, ::MonitorMixin)
         end
+
+        attr_accessor :_cancellation_error_class
       end
 
       # A monitor-owned condition variable composed with a (sortable) value.
@@ -69,7 +71,10 @@ def synchronized_take(n_items_argument = nil, timeout = nil, policy: :partial_on
         end
 
         synchronize do
-          if size >= n_items
+          if (cancellation_error_class = _cancellation_error_class)
+            # Canceled? Bail.
+            raise cancellation_error_class
+          elsif size >= n_items
             # If there happen to be `n_items`, just take and return.
             return block.call(n_items_argument ? n_items : nil)
           elsif policy == :partial && size > 0
@@ -84,7 +89,13 @@ def synchronized_take(n_items_argument = nil, timeout = nil, policy: :partial_on
           loop do
             vcv.wait(timeout)
 
-            if size >= n_items
+            if (cancellation_error_class = _cancellation_error_class)
+              # Deregister the taker because the operation was cancelled.
+              taker_priority_queue.delete(vcv)
+
+              # Bail after doing the above bookkeeping.
+              raise cancellation_error_class
+            elsif size >= n_items
               # Deregister the taker because we got `n_items`.
               taker_priority_queue.delete(vcv)
 
@@ -139,6 +150,28 @@ def signal_takers
           vcv.signal
         end
       end
+
+      # Cancel all outstanding synchronized operations.
+      def cancel(error_class)
+        if !error_class.try(:<, StandardError)
+          raise ArgumentError, "Please provide an error class"
+        end
+
+        synchronize do
+          while (vcv = taker_priority_queue.pop)
+            vcv.signal
+          end
+
+          self._cancellation_error_class = error_class
+        end
+      end
+
+      # Unset the synchronize operation cancellation status.
+      def uncancel
+        synchronize do
+          self._cancellation_error_class = nil
+        end
+      end
     end
   end
 end

spec/concurrent/concurrent_array_spec.rb

@@ -111,4 +111,27 @@
     expect(items_4).to eq([0, 1, 2, 3, 4])
     expect(items_5).to eq([0, 1, 2, 3, 4])
   end
+
+  it "cancels outstanding synchronized operations" do
+    shared_queue = ConcurrentArray.new
+
+    caught_error = nil
+
+    receiver = Thread.new do
+      begin
+        shared_queue.s_shift
+      rescue ::Concurrent::CancelledOperationError => e
+        caught_error = e
+      end
+    end
+
+    # Allow some time for the thread to enter condition variable `wait`.
+    sleep(0.1)
+
+    shared_queue.cancel(::Concurrent::CancelledOperationError)
+
+    receiver.join
+
+    expect(caught_error).to be_a(::Concurrent::CancelledOperationError)
+  end
 end