Operations can't timeout when writing to the write buffer.

We implemented a mechanism for operations to be able to timeout
while waiting to be written to a socket, but failed to take into
account the fact that an item can timeout while in the middle
of writing to that buffer.

This commit adds a new OperationState called WRITE_QUEUED that
specifies that the operation is waiting to be written to the
write buffer. When an operation begins writing to the write
buffer it goes into the WRITING state. When an operation is in
the WRITING state it cannot time out

Change-Id: Ib4ed9e1764c31e9270f79206b11574d98638e989
Reviewed-on: http://review.couchbase.org/9424
Reviewed-by: Michael Wiederhold <[email protected]>
Tested-by: Michael Wiederhold <[email protected]>

Author: Mike Wiederhold

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

@@ -1941,7 +1941,7 @@ public boolean cancel(boolean ign) {
 				boolean rv=false;
 				for(Operation op : ops) {
 					op.cancel();
-					rv |= op.getState() == OperationState.WRITING;
+					rv |= op.getState() == OperationState.WRITE_QUEUED;
 				}
 				return rv;
 			}

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

@@ -43,7 +43,7 @@ public BulkGetFuture(Map<String, Future<T>> m,
 	public boolean cancel(boolean ign) {
 		boolean rv=false;
 		for(Operation op : ops) {
-			rv |= op.getState() == OperationState.WRITING;
+			rv |= op.getState() == OperationState.WRITE_QUEUED;
 			op.cancel();
 		}
 		for (Future<T> v : rvMap.values()) {
@@ -174,4 +174,4 @@ public boolean isDone() {
 	public boolean isTimeout() {
 		return timeout;
 	}
-}
+}

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

@@ -48,7 +48,7 @@ public boolean cancel(boolean ign) {
 		op.cancel();
 		// This isn't exactly correct, but it's close enough.  If we're in
 		// a writing state, we *probably* haven't started.
-		return op.getState() == OperationState.WRITING;
+		return op.getState() == OperationState.WRITE_QUEUED;
 	}
 
 	public T get() throws InterruptedException, ExecutionException {

src/main/java/net/spy/memcached/ops/BaseOperationFactory.java

@@ -21,7 +21,7 @@ private String first(Collection<String> keys) {
 	}
 
 	public Collection<Operation> clone(KeyedOperation op) {
-		assert (op.getState() == OperationState.WRITING || op.getState() == OperationState.RETRY)
+		assert (op.getState() == OperationState.WRITE_QUEUED || op.getState() == OperationState.RETRY)
 			: "Who passed me an operation in the " + op.getState() + "state?";
 		assert !op.isCancelled() : "Attempted to clone a canceled op";
 		assert !op.hasErrored() : "Attempted to clone an errored op";

src/main/java/net/spy/memcached/ops/Operation.java

@@ -46,6 +46,12 @@ public interface Operation {
 	 */
 	ByteBuffer getBuffer();
 
+	/**
+	 * Invoked when we start writing all of the bytes from this operation to
+	 * the sockets write buffer.
+	 */
+	void writing();
+
 	/**
 	 * Invoked after having written all of the bytes from the supplied output
 	 * buffer.

src/main/java/net/spy/memcached/ops/OperationState.java

@@ -4,6 +4,10 @@
  * State of this operation.
  */
 public enum OperationState {
+	/**
+	 * State indicating this operation is waiting to be written to the server.
+	 */
+	WRITE_QUEUED,
 	/**
 	 * State indicating this operation is writing data to the server.
 	 */

src/main/java/net/spy/memcached/protocol/BaseOperationImpl.java

@@ -25,7 +25,7 @@ public abstract class BaseOperationImpl extends SpyObject implements Operation {
 	 */
 	public static final OperationStatus CANCELLED =
 		new CancelledOperationStatus();
-	private OperationState state = OperationState.WRITING;
+	private OperationState state = OperationState.WRITE_QUEUED;
 	private ByteBuffer cmd = null;
 	private boolean cancelled = false;
 	private OperationException exception = null;
@@ -104,7 +104,7 @@ protected final void transitionState(OperationState newState) {
 		getLogger().debug("Transitioned state from %s to %s", state, newState);
 		state=newState;
 		// Discard our buffer when we no longer need it.
-		if(state != OperationState.WRITING) {
+		if(state != OperationState.WRITE_QUEUED && state != OperationState.WRITING) {
 			cmd=null;
 		}
 		if(state == OperationState.COMPLETE) {
@@ -116,6 +116,10 @@ protected final void transitionState(OperationState newState) {
 		}
 	}
 
+	public final void writing() {
+		transitionState(OperationState.WRITING);
+	}
+
 	public final void writeComplete() {
 		transitionState(OperationState.READING);
 	}

src/main/java/net/spy/memcached/protocol/TCPMemcachedNodeImpl.java

@@ -157,17 +157,21 @@ public final void fillWriteBuffer(boolean shouldOptimize) {
 		if(toWrite == 0 && readQ.remainingCapacity() > 0) {
 			getWbuf().clear();
 			Operation o=getCurrentWriteOp();
-			if (o != null && (o.isCancelled())) {
-				getLogger().debug("Not writing cancelled op.");
-				Operation cancelledOp = removeCurrentWriteOp();
-				assert o == cancelledOp;
-				return;
-                        }
-			if (o != null && o.isTimedOut(defaultOpTimeout)) {
-				getLogger().debug("Not writing timed out op.");
-				Operation timedOutOp = removeCurrentWriteOp();
-				assert o == timedOutOp;
-				return;
+			
+			if (o != null && o.getState() == OperationState.WRITE_QUEUED) {
+				if (o.isCancelled()) {
+					getLogger().debug("Not writing cancelled op.");
+					Operation cancelledOp = removeCurrentWriteOp();
+					assert o == cancelledOp;
+					return;
+				} else if (o.isTimedOut(defaultOpTimeout)) {
+					getLogger().debug("Not writing timed out op.");
+					Operation timedOutOp = removeCurrentWriteOp();
+					assert o == timedOutOp;
+					return;
+				} else {
+					o.writing();
+				}
 			}
 			while(o != null && toWrite < getWbuf().capacity()) {
 				assert o.getState() == OperationState.WRITING;
@@ -199,6 +203,9 @@ public final void fillWriteBuffer(boolean shouldOptimize) {
 					}
 
 					o=getCurrentWriteOp();
+					if (o != null) {
+						o.writing();
+					}
 				}
 				toWrite += bytesToCopy;
 			}
@@ -517,5 +524,4 @@ public final void setupForAuth() {
 			authLatch = new CountDownLatch(0);
 		}
 	}
-
 }

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

@@ -41,12 +41,11 @@ protected void optimize() {
 
 				// Initialize the new mega get
 				optimizedOp.initialize();
-				assert optimizedOp.getState() == OperationState.WRITING;
+				assert optimizedOp.getState() == OperationState.WRITE_QUEUED;
 				ProxyCallback pcb=(ProxyCallback) og.getCallback();
 				getLogger().debug("Set up %s with %s keys and %s callbacks",
 					this, pcb.numKeys(), pcb.numCallbacks());
 			}
 		}
 	}
-
 }

src/main/java/net/spy/memcached/protocol/binary/BinaryMemcachedNodeImpl.java

@@ -56,7 +56,7 @@ private void optimizeGets() {
 
 			// Initialize the new mega get
 			optimizedOp.initialize();
-			assert optimizedOp.getState() == OperationState.WRITING;
+			assert optimizedOp.getState() == OperationState.WRITE_QUEUED;
 			ProxyCallback pcb=(ProxyCallback) og.getCallback();
 			getLogger().debug("Set up %s with %s keys and %s callbacks",
 					this, pcb.numKeys(), pcb.numCallbacks());
@@ -83,7 +83,7 @@ private void optimizeSets() {
 
 			// Initialize the new mega set
 			optimizedOp.initialize();
-			assert optimizedOp.getState() == OperationState.WRITING;
+			assert optimizedOp.getState() == OperationState.WRITE_QUEUED;
 		}
 	}
 }