Explicit support for acquire_semaphore/release_semaphore in semian re…

…sources

ext/semian/resource.c

@@ -5,9 +5,6 @@ ID id_wait_time;
 ID id_timeout;
 int system_max_semaphore_count;
 
-static VALUE
-cleanup_semian_resource_acquire(VALUE self);
-
 static void
 check_tickets_xor_quota_arg(VALUE tickets, VALUE quota);
 
@@ -33,15 +30,11 @@ static const rb_data_type_t
 semian_resource_type;
 
 VALUE
-semian_resource_acquire(int argc, VALUE *argv, VALUE self)
+semian_resource_acquire_semaphore(int argc, VALUE *argv, VALUE self)
 {
   semian_resource_t *self_res = NULL;
   semian_resource_t res = { 0 };
 
-  if (!rb_block_given_p()) {
-    rb_raise(rb_eArgError, "acquire requires a block");
-  }
-
   TypedData_Get_Struct(self, semian_resource_t, &semian_resource_type, self_res);
   res = *self_res;
 
@@ -75,7 +68,19 @@ semian_resource_acquire(int argc, VALUE *argv, VALUE self)
     wait_time = LONG2NUM(res.wait_time);
   }
 
-  return rb_ensure(rb_yield, wait_time, cleanup_semian_resource_acquire, self);
+  return wait_time;
+}
+
+VALUE
+semian_resource_acquire(int argc, VALUE *argv, VALUE self)
+{
+  if (!rb_block_given_p()) {
+    rb_raise(rb_eArgError, "acquire requires a block");
+  }
+
+  VALUE wait_time = semian_resource_acquire_semaphore(argc, argv, self);
+
+  return rb_ensure(rb_yield, wait_time, semian_resource_release_semaphore, self);
 }
 
 VALUE
@@ -253,8 +258,8 @@ semian_resource_in_use(VALUE self)
   return Qtrue;
 }
 
-static VALUE
-cleanup_semian_resource_acquire(VALUE self)
+VALUE
+semian_resource_release_semaphore(VALUE self)
 {
   semian_resource_t *res = NULL;
   TypedData_Get_Struct(self, semian_resource_t, &semian_resource_type, res);

ext/semian/resource.h

@@ -33,10 +33,39 @@ semian_resource_initialize(VALUE self, VALUE id, VALUE tickets, VALUE quota, VAL
  *
  * If no timeout argument is provided, the default timeout passed to Semian.register will be used.
  *
+ * The given block is executed with the semaphore held and, when the block
+ * exits, the semaphore is automatically released.
  */
 VALUE
 semian_resource_acquire(int argc, VALUE *argv, VALUE self);
 
+/*
+ * call-seq:
+ *    resource.acquire_semaphore(timeout: default_timeout) -> wait_time
+ *
+ * Acquires a resource. The call will block for <code>timeout</code> seconds if a ticket
+ * is not available. If no ticket is available within the timeout period, Semian::TimeoutError
+ * will be raised.
+ *
+ * If no timeout argument is provided, the default timeout passed to Semian.register will be used.
+ *
+ * Note: The caller is responsible for releasing the semaphore when done by calling release_semaphore.
+ */
+VALUE
+semian_resource_acquire_semaphore(int argc, VALUE *argv, VALUE self);
+
+/*
+ * call-seq:
+ *    resource.release_semaphore() -> nil
+ *
+ * Releases a resource previously acquired with acquire_semaphore.
+ *
+ * Note: The method is NOT idempotent. The caller must ensure that the method is called exactly
+ * as many times as acquire_semaphore.
+ */
+VALUE
+semian_resource_release_semaphore(VALUE self);
+
 /*
  * call-seq:
  *   resource.destroy() -> true

ext/semian/semian.c

@@ -60,6 +60,8 @@ void Init_semian()
   rb_define_alloc_func(cResource, semian_resource_alloc);
   rb_define_method(cResource, "initialize_semaphore", semian_resource_initialize, 5);
   rb_define_method(cResource, "acquire", semian_resource_acquire, -1);
+  rb_define_method(cResource, "acquire_semaphore", semian_resource_acquire_semaphore, -1);
+  rb_define_method(cResource, "release_semaphore", semian_resource_release_semaphore, 0);
   rb_define_method(cResource, "count", semian_resource_count, 0);
   rb_define_method(cResource, "semid", semian_resource_id, 0);
   rb_define_method(cResource, "key", semian_resource_key, 0);

lib/semian/resource.rb

@@ -46,6 +46,13 @@ def initialize(name, tickets: nil, quota: nil, permissions: Semian.default_permi
       yield wait_time
     end
 
+    redefinable def acquire_semaphore
+      0
+    end
+
+    redefinable def release_semaphore
+    end
+
     redefinable def count
       0
     end

test/resource_test.rb

@@ -140,6 +140,22 @@ def test_acquire
     assert(acquired)
   end
 
+  def test_acquire_and_release_semaphore
+    resource = create_resource(:testing, tickets: 1)
+
+    assert_equal(1, resource.count)
+
+    begin
+      resource.acquire_semaphore
+
+      assert_equal(0, resource.count)
+    ensure
+      resource.release_semaphore
+
+      assert_equal(1, resource.count)
+    end
+  end
+
   def test_acquire_return_val
     resource = create_resource(:testing, tickets: 1)
     val = resource.acquire { 1234 }
@@ -155,6 +171,38 @@ def test_acquire_timeout
     assert_equal(1, timeouts)
   end
 
+  def test_acquire_and_release_timeout
+    resource = create_resource(:testing, tickets: 1, timeout: 1)
+    start_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+    # Grab the only ticket
+    resource.acquire_semaphore
+
+    # Wait for the timeout trying to get a ticket, then catch the exception
+    assert_raises(Semian::TimeoutError) do
+      resource.acquire_semaphore
+    end
+    end_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+    # Should work now
+    resource.release_semaphore
+
+    # Ensure we actually waited for the timeout
+    assert_in_delta(1, end_time - start_time, EPSILON)
+  end
+
+  def test_release_semaphore_without_acquire
+    resource = create_resource(:testing, tickets: 1)
+
+    assert_equal(1, resource.count)
+
+    # Should not raise an error
+    resource.release_semaphore
+
+    # It has now screwed up the semaphore count because we didn't acquire before releasing
+    assert_equal(2, resource.count)
+  end
+
   def test_acquire_timeout_override
     skip("Never tested correctly")