Allow semaphore acquisition with blocks

Ruby often idomaticaly uses blocks to open a resource and close it
again automatically. For example:

    file = File.open('file', 'r')
    # do work with file
    file.close

    # vs

    File.open('file', 'r') do |file|
      # do work with file
    end
    # file closed automatically

This teaches semaphores a similar technique:

    semaphore = Concurrent::Semaphore.new(1)

    semaphore.acquire
    # do work
    semaphore.release

    # vs

    semaphore.acquire do
      # do work
    end
    # permit automatically released

Author: sambostock

ext/concurrent-ruby/com/concurrent_ruby/ext/JavaSemaphoreLibrary.java

@@ -10,6 +10,7 @@
 import org.jruby.RubyObject;
 import org.jruby.anno.JRubyClass;
 import org.jruby.anno.JRubyMethod;
+import org.jruby.runtime.Block;
 import org.jruby.runtime.ObjectAllocator;
 import org.jruby.runtime.ThreadContext;
 import org.jruby.runtime.builtin.IRubyObject;
@@ -45,13 +46,13 @@ public IRubyObject initialize(ThreadContext context, IRubyObject value) {
         }
 
         @JRubyMethod
-        public IRubyObject acquire(ThreadContext context) throws InterruptedException {
-            return this.acquire(context, 1);
+        public IRubyObject acquire(ThreadContext context, final Block block) throws InterruptedException {
+            return this.acquire(context, 1, block);
         }
 
         @JRubyMethod
-        public IRubyObject acquire(ThreadContext context, IRubyObject permits) throws InterruptedException {
-            return this.acquire(context, rubyFixnumToPositiveInt(permits, "permits"));
+        public IRubyObject acquire(ThreadContext context, IRubyObject permits, final Block block) throws InterruptedException {
+            return this.acquire(context, rubyFixnumToPositiveInt(permits, "permits"), block);
         }
 
         @JRubyMethod(name = "available_permits")
@@ -65,31 +66,31 @@ public IRubyObject drainPermits(ThreadContext context) {
         }
 
         @JRubyMethod(name = "try_acquire")
-        public IRubyObject tryAcquire(ThreadContext context) throws InterruptedException {
+        public IRubyObject tryAcquire(ThreadContext context, final Block block) throws InterruptedException {
             int permitsInt = 1;
             boolean acquired = semaphore.tryAcquire(permitsInt);
 
-            return triedAcquire(context, acquired);
+            return triedAcquire(context, permitsInt, acquired, block);
         }
 
         @JRubyMethod(name = "try_acquire")
-        public IRubyObject tryAcquire(ThreadContext context, IRubyObject permits) throws InterruptedException {
+        public IRubyObject tryAcquire(ThreadContext context, IRubyObject permits, final Block block) throws InterruptedException {
             int permitsInt = rubyFixnumToPositiveInt(permits, "permits");
             boolean acquired = semaphore.tryAcquire(permitsInt);
 
-            return triedAcquire(context, acquired);
+            return triedAcquire(context, permitsInt, acquired, block);
         }
 
         @JRubyMethod(name = "try_acquire")
-        public IRubyObject tryAcquire(ThreadContext context, IRubyObject permits, IRubyObject timeout) throws InterruptedException {
+        public IRubyObject tryAcquire(ThreadContext context, IRubyObject permits, IRubyObject timeout, final Block block) throws InterruptedException {
             int permitsInt = rubyFixnumToPositiveInt(permits, "permits");
             boolean acquired = semaphore.tryAcquire(
                     permitsInt,
                     rubyNumericToLong(timeout, "timeout"),
-                    java.util.concurrent.TimeUnit.SECONDS,
+                    java.util.concurrent.TimeUnit.SECONDS
                     );
 
-            return triedAcquire(context, acquired);
+            return triedAcquire(context, permitsInt, acquired, block);
         }
 
         @JRubyMethod
@@ -110,14 +111,27 @@ public IRubyObject reducePermits(ThreadContext context, IRubyObject reduction) t
             return context.nil;
         }
 
-        private IRubyObject acquire(ThreadContext context, int permits) throws InterruptedException {
+        private IRubyObject acquire(ThreadContext context, int permits, final Block block) throws InterruptedException {
             this.semaphore.acquire(permits);
 
-            return context.nil;
+            if (!block.isGiven()) return context.nil;
+
+            try {
+                return block.yieldSpecific(context);
+            } finally {
+                this.semaphore.release(permits);
+            }
         }
 
-        private IRubyObject triedAcquire(ThreadContext context, int permits, boolean acquired) {
-            return getRuntime().newBoolean(acquired);
+        private IRubyObject triedAcquire(ThreadContext context, int permits, boolean acquired, final Block block) {
+            if (!block.isGiven()) return getRuntime().newBoolean(acquired);
+            if (!acquired) return context.nil;
+
+            try {
+                return block.yieldSpecific(context);
+            } finally {
+                this.semaphore.release(permits);
+            }
         }
 
         private int rubyFixnumInt(IRubyObject value, String paramName) {

lib/concurrent-ruby/concurrent/atomic/mutex_semaphore.rb

@@ -25,7 +25,13 @@ def acquire(permits = 1)
         try_acquire_timed(permits, nil)
       end
 
-      nil
+      return unless block_given?
+
+      begin
+        yield
+      ensure
+        release(permits)
+      end
     end
 
     # @!macro semaphore_method_available_permits
@@ -49,13 +55,22 @@ def try_acquire(permits = 1, timeout = nil)
       Utility::NativeInteger.ensure_integer_and_bounds permits
       Utility::NativeInteger.ensure_positive permits
 
-      synchronize do
+      acquired = synchronize do
         if timeout.nil?
           try_acquire_now(permits)
         else
           try_acquire_timed(permits, timeout)
         end
       end
+
+      return acquired unless block_given?
+      return unless acquired
+
+      begin
+        yield
+      ensure
+        release(permits)
+      end
     end
 
     # @!macro semaphore_method_release

lib/concurrent-ruby/concurrent/atomic/semaphore.rb

@@ -16,14 +16,16 @@ module Concurrent
   # @!macro semaphore_method_acquire
   #
   #   Acquires the given number of permits from this semaphore,
-  #     blocking until all are available.
+  #     blocking until all are available. If a block is given,
+  #     yields to it and releases the permits afterwards.
   #
   #   @param [Fixnum] permits Number of permits to acquire
   #
   #   @raise [ArgumentError] if `permits` is not an integer or is less than
   #     one
   #
-  #   @return [nil]
+  #   @return [nil, BasicObject] Without a block, `nil` is returned. If a block
+  #   is given, its return value is returned.
 
   # @!macro semaphore_method_available_permits
   #
@@ -41,7 +43,9 @@ module Concurrent
   #
   #   Acquires the given number of permits from this semaphore,
   #     only if all are available at the time of invocation or within
-  #     `timeout` interval
+  #     `timeout` interval. If a block is given, yields to it if the permits
+  #     were successfully acquired, and releases them afterward, returning the
+  #     block's return value.
   #
   #   @param [Fixnum] permits the number of permits to acquire
   #
@@ -51,8 +55,10 @@ module Concurrent
   #   @raise [ArgumentError] if `permits` is not an integer or is less than
   #     one
   #
-  #   @return [Boolean] `false` if no permits are available, `true` when
-  #     acquired a permit
+  #   @return [true, false, nil, BasicObject] `false` if no permits are
+  #     available, `true` when acquired a permit. If a block is given, the
+  #     block's return value is returned if the permits were acquired; if not,
+  #     `nil` is returned.
 
   # @!macro semaphore_method_release
   #
@@ -106,6 +112,8 @@ module Concurrent
   #   releasing a blocking acquirer.
   #   However, no actual permit objects are used; the Semaphore just keeps a
   #   count of the number available and acts accordingly.
+  #   Alternatively, permits may be acquired within a block, and automatically
+  #   released after the block finishes executing.
   #
   # @!macro semaphore_public_api
   # @example
@@ -140,6 +148,19 @@ module Concurrent
   #   # Thread 4 releasing semaphore
   #   # Thread 1 acquired semaphore
   #
+  # @example
+  #   semaphore = Concurrent::Semaphore.new(1)
+  #
+  #   puts semaphore.available_permits
+  #   semaphore.acquire do
+  #     puts semaphore.available_permits
+  #   end
+  #   puts semaphore.available_permits
+  #
+  #   # prints:
+  #   # 1
+  #   # 0
+  #   # 1
   class Semaphore < SemaphoreImplementation
   end
 end