INTERNAL: make piped insert operations process synchronously

src/main/java/net/spy/memcached/ArcusClient.java

@@ -38,6 +38,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
 import java.util.jar.JarFile;
 import java.util.jar.Manifest;
 
@@ -1783,7 +1784,7 @@ public <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> asyncBopPip
         insertList.add(new BTreePipedInsert<>(key, elementMap, attributesForCreate, tc));
       }
     }
-    return asyncCollectionPipedInsert(key, insertList);
+    return syncCollectionPipedInsert(key, insertList);
   }
 
   @Override
@@ -1806,7 +1807,7 @@ public <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> asyncBopPip
         insertList.add(new ByteArraysBTreePipedInsert<>(key, elementList, attributesForCreate, tc));
       }
     }
-    return asyncCollectionPipedInsert(key, insertList);
+    return syncCollectionPipedInsert(key, insertList);
   }
 
   @Override
@@ -1833,7 +1834,7 @@ public <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> asyncMopPip
         insertList.add(new MapPipedInsert<>(key, elementMap, attributesForCreate, tc));
       }
     }
-    return asyncCollectionPipedInsert(key, insertList);
+    return syncCollectionPipedInsert(key, insertList);
   }
 
   @Override
@@ -1859,7 +1860,7 @@ public <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> asyncLopPip
         }
       }
     }
-    return asyncCollectionPipedInsert(key, insertList);
+    return syncCollectionPipedInsert(key, insertList);
   }
 
   @Override
@@ -1882,7 +1883,7 @@ public <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> asyncSopPip
         insertList.add(new SetPipedInsert<>(key, elementList, attributesForCreate, tc));
       }
     }
-    return asyncCollectionPipedInsert(key, insertList);
+    return syncCollectionPipedInsert(key, insertList);
   }
 
   @Override
@@ -3106,6 +3107,78 @@ public void gotStatus(Integer index, OperationStatus status) {
     return rv;
   }
 
+  /**
+   * insert items into collection synchronously.
+   *
+   * @param key arcus cache key
+   * @param insertList must not be empty.
+   * @return future holding the map of element index and the reason why insert operation failed
+   */
+  private <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> syncCollectionPipedInsert(
+          final String key, final List<CollectionPipedInsert<T>> insertList) {
+    final CountDownLatch latch = new CountDownLatch(1);
+    final PipedCollectionFuture<Integer, CollectionOperationStatus> rv =
+            new PipedCollectionFuture<>(latch, operationTimeout);
+
+    BiFunction<Integer, Integer, OperationCallback> makeCallback = (opIdx, itemCount) -> new PipedOperationCallback() {
+
+      private int currentItemIdx = -1;
+
+      public void receivedStatus(OperationStatus status) {
+        CollectionOperationStatus cstatus;
+
+        if (status instanceof CollectionOperationStatus) {
+          cstatus = (CollectionOperationStatus) status;
+        } else {
+          getLogger().warn("Unhandled state: " + status);
+          cstatus = new CollectionOperationStatus(status);
+        }
+        rv.setOperationStatus(cstatus);
+      }
+
+      public void complete() {
+        if (rv.getOperationStatus().isSuccess()) {
+          Operation nextOp = rv.getNextOp();
+          if (nextOp != null && !nextOp.isCancelled()) {
+            addOp(key, nextOp);
+            rv.incrCurrentOpIdx();
+            return;
+          }
+        } else {
+          // If this operation has been errored or cancelled, some items in the operation may not be executed.
+          // The first item that is not executed will have the STOPPED flag.
+          if ((currentItemIdx + 1) < itemCount || (opIdx + 1 < insertList.size())) {
+            this.gotStatus(currentItemIdx + 1,
+                    new CollectionOperationStatus(false, "STOPPED", CollectionResponse.STOPPED));
+          }
+        }
+        latch.countDown();
+      }
+
+      public void gotStatus(Integer index, OperationStatus status) {
+        if (!status.isSuccess()) {
+          if (status instanceof CollectionOperationStatus) {
+            rv.addEachResult(index + (opIdx * MAX_PIPED_ITEM_COUNT),
+                    (CollectionOperationStatus) status);
+          } else {
+            rv.addEachResult(index + (opIdx * MAX_PIPED_ITEM_COUNT),
+                    new CollectionOperationStatus(status));
+          }
+        }
+        currentItemIdx = index;
+      }
+    };
+
+    for (int i = 0; i < insertList.size(); i++) {
+      final CollectionPipedInsert<T> insert = insertList.get(i);
+      Operation op = opFact.collectionPipedInsert(key, insert, makeCallback.apply(i, insert.getItemCount()));
+      rv.addOperation(op);
+    }
+    addOp(key, rv.getOperation(0));
+    rv.incrCurrentOpIdx();
+    return rv;
+  }
+
   @Override
   public Future<Map<String, CollectionOperationStatus>> asyncBopInsertBulk(
           List<String> keyList, long bkey, byte[] eFlag, Object value,

src/main/java/net/spy/memcached/collection/CollectionResponse.java

@@ -44,6 +44,7 @@ public enum CollectionResponse {
 
   UNDEFINED,
   CANCELED,
+  STOPPED,
 
   INTERRUPT_EXCEPTION,
   EXECUTION_EXCEPTION,

src/main/java/net/spy/memcached/internal/PipedCollectionFuture.java

@@ -2,6 +2,7 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
@@ -17,9 +18,10 @@
 
 public class PipedCollectionFuture<K, V>
         extends CollectionFuture<Map<K, V>> {
-  private final Collection<Operation> ops = new ArrayList<>();
+  private final List<Operation> ops = new ArrayList<>();
   private final AtomicReference<CollectionOperationStatus> operationStatus
           = new AtomicReference<>(null);
+  private int currentOpIdx = -1;
 
   private final Map<K, V> failedResult =
           new ConcurrentHashMap<>();
@@ -30,17 +32,21 @@ public PipedCollectionFuture(CountDownLatch l, long opTimeout) {
 
   @Override
   public boolean cancel(boolean ign) {
-    boolean rv = false;
-    for (Operation op : ops) {
-      rv |= op.cancel("by application.");
+    for (int i = currentOpIdx; i < ops.size(); i++) {
+      if (ops.get(i).cancel("by application.")) {
+        return true;
+      }
     }
-    return rv;
+    return false;
   }
 
+  /**
+   * @return true if any operation is cancelled.
+   */
   @Override
   public boolean isCancelled() {
-    for (Operation op : ops) {
-      if (op.isCancelled()) {
+    for (int i = currentOpIdx; i < ops.size(); i++) {
+      if (ops.get(i).isCancelled()) {
         return true;
       }
     }
@@ -119,4 +125,20 @@ public void addEachResult(K index, V status) {
   public void addOperation(Operation op) {
     ops.add(op);
   }
+
+  public Operation getOperation(int idx) {
+    return ops.get(idx);
+  }
+
+  public void incrCurrentOpIdx() {
+    this.currentOpIdx++;
+  }
+
+  public Operation getNextOp() {
+    try {
+      return ops.get(currentOpIdx + 1);
+    } catch (IndexOutOfBoundsException e) {
+      return null;
+    }
+  }
 }

src/main/java/net/spy/memcached/protocol/ascii/CollectionPipedInsertOperationImpl.java

@@ -29,7 +29,7 @@
 /**
  * Operation to store collection data in a memcached server.
  */
-public final class CollectionPipedInsertOperationImpl extends PipeOperationImpl
+public final class CollectionPipedInsertOperationImpl extends SingleKeyPipeOperationImpl
         implements CollectionPipedInsertOperation {
 
   public CollectionPipedInsertOperationImpl(String key,

src/main/java/net/spy/memcached/protocol/ascii/PipeOperationImpl.java

@@ -62,13 +62,13 @@ abstract class PipeOperationImpl extends OperationImpl {
 
   protected boolean successAll = true;
 
-  private final CollectionPipe collectionPipe;
-  private final PipedOperationCallback cb;
-  private final List<String> keys;
+  protected final CollectionPipe collectionPipe;
+  protected final PipedOperationCallback cb;
+  protected final List<String> keys;
   private final boolean isIdempotent;
 
-  private int index = 0;
-  private boolean readUntilLastLine = false;
+  protected int index = 0;
+  protected boolean readUntilLastLine = false;
 
   protected PipeOperationImpl(List<String> keys, CollectionPipe collectionPipe,
                               OperationCallback cb) {

src/main/java/net/spy/memcached/protocol/ascii/SingleKeyPipeOperationImpl.java

@@ -0,0 +1,99 @@
+package net.spy.memcached.protocol.ascii;
+
+import java.util.List;
+
+import net.spy.memcached.collection.CollectionPipe;
+import net.spy.memcached.ops.OperationCallback;
+import net.spy.memcached.ops.OperationState;
+import net.spy.memcached.ops.OperationStatus;
+
+public abstract class SingleKeyPipeOperationImpl extends PipeOperationImpl {
+
+  protected SingleKeyPipeOperationImpl(List<String> keys,
+                                       CollectionPipe collectionPipe,
+                                       OperationCallback cb) {
+    super(keys, collectionPipe, cb);
+  }
+
+  @Override
+  public void handleLine(String line) {
+    assert getState() == OperationState.READING
+            : "Read ``" + line + "'' when in " + getState() + " state";
+
+    /* ENABLE_REPLICATION if */
+    if (isWriteOperation() && hasSwitchedOver(line)) {
+      collectionPipe.setNextOpIndex(index);
+      prepareSwitchover(line);
+      return;
+    }
+    /* ENABLE_REPLICATION end */
+
+    /* ENABLE_MIGRATION if */
+    if (hasNotMyKey(line)) {
+      // Only one NOT_MY_KEY is provided in response of
+      // single key piped operation when redirection.
+      addRedirectSingleKeyOperation(line, keys.get(0));
+      if (collectionPipe.isNotPiped()) {
+        transitionState(OperationState.REDIRECT);
+      } else {
+        collectionPipe.setNextOpIndex(index);
+      }
+      return;
+    }
+    /* ENABLE_MIGRATION end */
+
+    if (collectionPipe.isNotPiped()) {
+      OperationStatus status = checkStatus(line);
+      if (!status.isSuccess()) {
+        successAll = false;
+      }
+      cb.gotStatus(index, status);
+
+      cb.receivedStatus((successAll) ? END : FAILED_END);
+      transitionState(OperationState.COMPLETE);
+      return;
+    }
+
+    /*
+      RESPONSE <count>\r\n
+      <status of the 1st pipelined command>\r\n
+      [ ... ]
+      <status of the last pipelined command>\r\n
+      END|PIPE_ERROR <error_string>\r\n
+    */
+    if (line.startsWith("END") || line.startsWith("PIPE_ERROR ")) {
+      /* ENABLE_MIGRATION if */
+      if (needRedirect()) {
+        transitionState(OperationState.REDIRECT);
+        return;
+      }
+      /* ENABLE_MIGRATION end */
+      cb.receivedStatus((index == collectionPipe.getItemCount() && successAll) ? END : FAILED_END);
+      transitionState(OperationState.COMPLETE);
+    } else if (line.startsWith("RESPONSE ")) {
+      getLogger().debug("Got line %s", line);
+
+      // TODO server should be fixed
+      line = line.replace("   ", " ");
+      line = line.replace("  ", " ");
+
+      String[] stuff = line.split(" ");
+      assert "RESPONSE".equals(stuff[0]);
+      readUntilLastLine = true;
+    } else {
+      OperationStatus status = checkStatus(line);
+      if (!status.isSuccess()) {
+        successAll = false;
+      }
+      cb.gotStatus(index, status);
+
+      index++;
+    }
+  }
+
+  @Override
+  protected OperationStatus checkStatus(String line) {
+    return null;
+  }
+
+}

src/test/manual/net/spy/memcached/bulkoperation/BopInsertBulkMultipleTest.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import net.spy.memcached.ArcusClient;
 import net.spy.memcached.collection.BaseIntegrationTest;
 import net.spy.memcached.collection.CollectionAttributes;
 import net.spy.memcached.collection.Element;
@@ -122,7 +123,7 @@ void testErrorCount() {
 
       Map<Integer, CollectionOperationStatus> map = future.get(2000L,
               TimeUnit.MILLISECONDS);
-      assertEquals(bkeySize, map.size());
+      assertEquals(ArcusClient.MAX_PIPED_ITEM_COUNT + 1, map.size());
 
     } catch (Exception e) {
       e.printStackTrace();

src/test/manual/net/spy/memcached/bulkoperation/LopInsertBulkMultipleValueTest.java

@@ -23,8 +23,10 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import net.spy.memcached.ArcusClient;
 import net.spy.memcached.collection.BaseIntegrationTest;
 import net.spy.memcached.collection.CollectionAttributes;
+import net.spy.memcached.collection.CollectionResponse;
 import net.spy.memcached.ops.CollectionOperationStatus;
 
 import org.junit.jupiter.api.AfterEach;
@@ -111,9 +113,7 @@ void testInsertAndGet() {
   void testErrorCount() {
     int valueCount = 1200;
     Object[] valueList = new Object[valueCount];
-    for (int i = 0; i < valueList.length; i++) {
-      valueList[i] = "MyValue";
-    }
+    Arrays.fill(valueList, "MyValue");
 
     try {
       // SET
@@ -123,8 +123,11 @@ void testErrorCount() {
 
       Map<Integer, CollectionOperationStatus> map = future.get(1000L,
               TimeUnit.MILLISECONDS);
-      assertEquals(valueCount, map.size());
-
+      assertEquals(ArcusClient.MAX_PIPED_ITEM_COUNT + 1, map.size());
+      assertEquals(map.get(ArcusClient.MAX_PIPED_ITEM_COUNT - 1).getResponse(),
+              CollectionResponse.NOT_FOUND);
+      assertEquals(map.get(ArcusClient.MAX_PIPED_ITEM_COUNT).getResponse(),
+              CollectionResponse.STOPPED);
     } catch (Exception e) {
       e.printStackTrace();
       fail();