Prefer to use a special RETRY_TOKEN to trigger retrying a semaphore acquire

Previously this would throw a RetryException, but this seems cleaner.
Throwing exceptions to control flow is typically not good, especially
because normal exceptions would also require filling in a stack trace
which adds additional resource consumption. In this case the
RetryException does not do that, but still this approach should be
better.

Author: jdeppe-pivotal

src/main/java/dev/gemfire/dtype/internal/AbstractDType.java

@@ -27,6 +27,8 @@
 
 public abstract class AbstractDType implements Delta, DataSerializable, DType {
 
+  static final Object RETRY_TOKEN = new Object();
+
   private String name;
   private transient Region<String, Object> region;
   private transient DTypeCollectionsFunction deltaOperation = null;

src/main/java/dev/gemfire/dtype/internal/CollectionsBackendFunction.java

@@ -22,15 +22,14 @@ public class CollectionsBackendFunction implements Function<Object> {
   public static final String ID = "dtype-collections-function";
 
   @Override
-  @SuppressWarnings("unchecked")
   public void execute(FunctionContext<Object> context) {
     Object[] args = (Object[]) context.getArguments();
     String name = (String) args[0];
     String memberTag = (String) args[1];
     DTypeCollectionsFunction fn = (DTypeCollectionsFunction) args[2];
     OperationType operationType = (OperationType) args[3];
 
-    Region<String, AbstractDType> region = ((RegionFunctionContext) context).getDataSet();
+    Region<String, AbstractDType> region = ((RegionFunctionContext<?>) context).getDataSet();
     AbstractDType entry = region.get(name);
 
     Callable<Object> wrappingFn = () -> {

src/main/java/dev/gemfire/dtype/internal/DSemaphoreBackend.java

@@ -68,16 +68,17 @@ public int getQueueLength() {
   }
 
   /**
-   * Acquire permits or throw an exception if unable to do so. The exception is processed in the
-   * {@link DSemaphoreBackendFunction} where the request is retried.
+   * Acquire permits or return a special 'retry token' if unable to do so. The
+   * token is processed in the {@link DSemaphoreBackendFunction} where the
+   * request is retried.
    *
    * @param context the relevant context for DSemaphore calls (DSemaphoreFunctionContext)
    * @param permits the number of permits to acquire
+   * @return null if successful or {@link AbstractDType#RETRY_TOKEN} if unable
+   * to acquire the requested permits
    */
-  public void acquire(DTypeFunctionContext context, int permits) {
-    if (!_acquire(context, permits)) {
-      throw new RetryableException(0);
-    }
+  public Object acquire(DTypeFunctionContext context, int permits) {
+    return _acquire(context, permits) ? null : AbstractDType.RETRY_TOKEN;
   }
 
   private boolean _acquire(DTypeFunctionContext context, int permits) {

src/main/java/dev/gemfire/dtype/internal/DSemaphoreBackendFunction.java

@@ -8,6 +8,7 @@
 
 import java.util.concurrent.Callable;
 
+import dev.gemfire.dtype.DType;
 import org.apache.logging.log4j.Logger;
 
 import org.apache.geode.cache.Region;
@@ -63,9 +64,10 @@ public void execute(FunctionContext<Object> context) {
       synchronized (finalEntry) {
         try {
           result = ((PartitionedRegion) region).computeWithPrimaryLocked(name, wrappingFn);
-          break;
-        } catch (RetryableException rex) {
-          // This exception is only thrown when an acquire() fails. We need to ensure that we wait()
+          if (result != AbstractDType.RETRY_TOKEN) {
+            break;
+          }
+          // This path is only reached when an acquire() fails. We need to ensure that we wait()
           // without holding the read lock on the region otherwise threads could get stuck. For
           // example, during bucket movement.
           if (logger.isDebugEnabled()) {

src/main/java/dev/gemfire/dtype/internal/DSemaphoreImpl.java

@@ -42,10 +42,7 @@ public void acquire() {
   @Override
   public void acquire(int permits) {
     validatePermits(permits);
-    DTypeContextualFunction fn = (sem, ctx) -> {
-      ((DSemaphoreBackend) sem).acquire(ctx, permits);
-      return null;
-    };
+    DTypeContextualFunction fn = (sem, ctx) -> ((DSemaphoreBackend) sem).acquire(ctx, permits);
     update(fn, DSemaphoreBackendFunction.ID);
   }