FIX: sync problem in BaseOperationImpl cancel and transitionState met…

…hod.
naver · Aug 1, 2023 · 98ed6ea · 98ed6ea
1 parent 6998801
commit 98ed6ea
Show file tree

Hide file tree

Showing 10 changed files with 20 additions and 40 deletions.
diff --git a/src/main/java/net/spy/memcached/ArcusClient.java b/src/main/java/net/spy/memcached/ArcusClient.java
@@ -1941,10 +1941,7 @@ public void complete() {
       public boolean cancel(boolean ign) {
         boolean rv = false;
         for (Operation op : ops) {
-          if (op.getState() != OperationState.COMPLETE) {
-            rv = true;
-            op.cancel("by application.");
-          }
+          rv |= op.cancel("by application.");
         }
         return rv;
       }

diff --git a/src/main/java/net/spy/memcached/MemcachedClient.java b/src/main/java/net/spy/memcached/MemcachedClient.java
@@ -2027,10 +2027,7 @@ public void complete() {
       public boolean cancel(boolean ign) {
         boolean rv = false;
         for (Operation op : ops) {
-          if (op.getState() != OperationState.COMPLETE) {
-            rv = true;
-            op.cancel("by application.");
-          }
+          rv |= op.cancel("by application.");
         }
         return rv;
       }

diff --git a/src/main/java/net/spy/memcached/internal/BulkGetFuture.java b/src/main/java/net/spy/memcached/internal/BulkGetFuture.java
@@ -64,10 +64,7 @@ public BulkGetFuture(BulkGetFuture<T> other) {
   public boolean cancel(boolean ign) {
     boolean rv = false;
     for (Operation op : ops) {
-      if (op.getState() != OperationState.COMPLETE) {
-        rv = true;
-        op.cancel("by application.");
-      }
+      rv |= op.cancel("by application.");
     }
     return rv;
   }

diff --git a/src/main/java/net/spy/memcached/internal/BulkOperationFuture.java b/src/main/java/net/spy/memcached/internal/BulkOperationFuture.java
@@ -31,10 +31,7 @@ public BulkOperationFuture(CountDownLatch l, long timeout) {
   public boolean cancel(boolean ign) {
     boolean rv = false;
     for (Operation op : ops) {
-      if (op.getState() != OperationState.COMPLETE) {
-        rv = true;
-        op.cancel("by application.");
-      }
+      rv |= op.cancel("by application.");
     }
     return rv;
   }

diff --git a/src/main/java/net/spy/memcached/internal/CollectionGetBulkFuture.java b/src/main/java/net/spy/memcached/internal/CollectionGetBulkFuture.java
@@ -92,10 +92,7 @@ public T get(long duration, TimeUnit units)
   public boolean cancel(boolean ign) {
     boolean rv = false;
     for (Operation op : ops) {
-      if (op.getState() != OperationState.COMPLETE) {
-        rv = true;
-        op.cancel("by application.");
-      }
+      rv |= op.cancel("by application.");
     }
     return rv;
   }

diff --git a/src/main/java/net/spy/memcached/internal/OperationFuture.java b/src/main/java/net/spy/memcached/internal/OperationFuture.java
@@ -60,11 +60,7 @@ public OperationFuture(CountDownLatch l, AtomicReference<T> oref,
 
   public boolean cancel(boolean ign) {
     assert op != null : "No operation";
-    if (op.getState() == OperationState.COMPLETE) {
-      return false;
-    }
-    op.cancel("by application.");
-    return true;
+    return op.cancel("by application.");
   }
 
   public T get() throws InterruptedException, ExecutionException {

diff --git a/src/main/java/net/spy/memcached/internal/PipedCollectionFuture.java b/src/main/java/net/spy/memcached/internal/PipedCollectionFuture.java
@@ -35,10 +35,7 @@ public PipedCollectionFuture(CountDownLatch l, long opTimeout) {
   public boolean cancel(boolean ign) {
     boolean rv = false;
     for (Operation op : ops) {
-      if (op.getState() != OperationState.COMPLETE) {
-        rv = true;
-        op.cancel("by application.");
-      }
+      rv |= op.cancel("by application.");
     }
     return rv;
   }

diff --git a/src/main/java/net/spy/memcached/internal/SMGetFuture.java b/src/main/java/net/spy/memcached/internal/SMGetFuture.java
@@ -53,10 +53,7 @@ public T get() throws InterruptedException, ExecutionException {
   public boolean cancel(boolean ign) {
     boolean rv = false;
     for (Operation op : ops) {
-      if (op.getState() != OperationState.COMPLETE) {
-        rv = true;
-        op.cancel("by application.");
-      }
+      rv |= op.cancel("by application.");
     }
     return rv;
   }

diff --git a/src/main/java/net/spy/memcached/ops/Operation.java b/src/main/java/net/spy/memcached/ops/Operation.java
@@ -52,7 +52,7 @@ public interface Operation {
   /**
    * Cancel this operation.
    */
-  void cancel(String cause);
+  boolean cancel(String cause);
 
   /**
    * Get the cause of cancel.

diff --git a/src/main/java/net/spy/memcached/protocol/BaseOperationImpl.java b/src/main/java/net/spy/memcached/protocol/BaseOperationImpl.java
@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import net.spy.memcached.MemcachedNode;
 import net.spy.memcached.MemcachedReplicaGroup;
@@ -45,9 +46,9 @@ public abstract class BaseOperationImpl extends SpyObject {
    */
   public static final OperationStatus CANCELLED =
           new CancelledOperationStatus();
-  private OperationState state = OperationState.WRITE_QUEUED;
+  private volatile OperationState state = OperationState.WRITE_QUEUED;
   private ByteBuffer cmd = null;
-  private boolean cancelled = false;
+  private AtomicBoolean cancelled = new AtomicBoolean(false);
   private String cancelCause = null;
   private OperationException exception = null;
   protected OperationCallback callback = null;
@@ -79,7 +80,7 @@ protected void setCallback(OperationCallback to) {
   }
 
   public final boolean isCancelled() {
-    return cancelled;
+    return cancelled.get();
   }
 
   public final boolean hasErrored() {
@@ -90,15 +91,19 @@ public final OperationException getException() {
     return exception;
   }
 
-  public final void cancel(String cause) {
-    cancelled = true;
+  public final boolean cancel(String cause) {
+    if (!cancelled.compareAndSet(false, true) ||
+            state == OperationState.COMPLETE) {
+      return false;
+    }
     if (handlingNode != null) {
       cancelCause = "Cancelled (" + cause + " : (" + handlingNode.getNodeName() + ")" + ")";
     } else {
       cancelCause = "Cancelled (" + cause + ")";
     }
     wasCancelled();
     callback.complete();
+    return true;
   }
 
   public final String getCancelCause() {
@@ -215,7 +220,7 @@ protected final void transitionState(OperationState newState) {
         state != OperationState.WRITING) {
       cmd = null;
     }
-    if (state == OperationState.COMPLETE) {
+    if (state == OperationState.COMPLETE && !this.cancelled.get()) {
       callback.complete();
     }
   }