Fix concurrent access to operations objects, especially near timeouts

We discovered that having one thread call get() on futures with a
timeout much shorter than the timeout built into the client could
(after a while) reliably kill off the IO thread with a
NullPointerException. After restarting with assertions enabled, we
found many other ways to make the IO thread die relating mostly to
operations that were canceled or timed out in one thread while they
were being used in another thread.

The solution to this is to make "timing out" an operation something
similar to "cancel" - that is, not a state of the operation's state
machine, but just a flag. Then, since this flag (as with the cancelled
flag) can be read and set from multiple threads, synchronize the
methods that do that.

Ideally, transitionState and getState would never be called outside
the IO thread; however, since at least getState is currently, both
getState and transitionState should then also be synchronized.

With these changes, we don't end up killing off the IO thread anymore.

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

Author: Daniel Martin

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

@@ -501,27 +501,27 @@ private void handleReads(SelectionKey sk, MemcachedNode qa)
 				if(currentOp == null) {
 					throw new IllegalStateException("No read operation.");
 				}
-				currentOp.readFromBuffer(rbuf);
-				if(currentOp.getState() == OperationState.COMPLETE) {
-					getLogger().debug(
-							"Completed read op: %s and giving the next %d bytes",
-							currentOp, rbuf.remaining());
-					Operation op=qa.removeCurrentReadOp();
-					assert op == currentOp
-					: "Expected to pop " + currentOp + " got " + op;
-					currentOp=qa.getCurrentReadOp();
-				} else if (currentOp.getState() == OperationState.RETRY) {
-                    getLogger().warn(
-                            "Reschedule read op due to NOT_MY_VBUCKET error: %s ",
-                            currentOp);
-                    ((VBucketAware) currentOp).addNotMyVbucketNode(currentOp.getHandlingNode());
-                    Operation op=qa.removeCurrentReadOp();
-                    assert op == currentOp
-                    : "Expected to pop " + currentOp + " got " + op;
-                    retryOps.add(currentOp);
-                    currentOp=qa.getCurrentReadOp();
-
+				synchronized(currentOp) {
+					currentOp.readFromBuffer(rbuf);
+					if(currentOp.getState() == OperationState.COMPLETE) {
+						getLogger().debug(
+								"Completed read op: %s and giving the next %d bytes",
+								currentOp, rbuf.remaining());
+						Operation op=qa.removeCurrentReadOp();
+						assert op == currentOp
+						: "Expected to pop " + currentOp + " got " + op;
+					} else if (currentOp.getState() == OperationState.RETRY) {
+						getLogger().warn(
+								"Reschedule read op due to NOT_MY_VBUCKET error: %s ",
+								currentOp);
+						((VBucketAware) currentOp).addNotMyVbucketNode(currentOp.getHandlingNode());
+						Operation op=qa.removeCurrentReadOp();
+						assert op == currentOp
+						: "Expected to pop " + currentOp + " got " + op;
+						retryOps.add(currentOp);
+					}
 				}
+				currentOp=qa.getCurrentReadOp();
 			}
 			rbuf.clear();
 			read=channel.read(rbuf);
@@ -597,6 +597,9 @@ private void cancelOperations(Collection<Operation> ops) {
 
 	private void redistributeOperations(Collection<Operation> ops) {
 		for(Operation op : ops) {
+			if (op.isCancelled() || op.isTimedOut()) {
+				continue;
+			}
 			if(op instanceof KeyedOperation) {
 				KeyedOperation ko = (KeyedOperation)op;
 				int added = 0;

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

@@ -20,10 +20,6 @@ public enum OperationState {
 	 * State indicating this operation is complete.
 	 */
 	COMPLETE,
-	/**
-	 * State indicating this operation timed out without completing.
-	 */
-	TIMEDOUT,
 	/**
 	 * State indicating this operation needs to be resent.  Typically
 	 * this means vbucket hashing and there is a topology change.

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

@@ -55,7 +55,7 @@ protected void setCallback(OperationCallback to) {
 		callback=to;
 	}
 
-	public final boolean isCancelled() {
+	public final synchronized boolean isCancelled() {
 		return cancelled;
 	}
 
@@ -67,7 +67,7 @@ public final OperationException getException() {
 		return exception;
 	}
 
-	public final void cancel() {
+	public final synchronized void cancel() {
 		cancelled=true;
 		wasCancelled();
 		callback.complete();
@@ -80,18 +80,18 @@ protected void wasCancelled() {
 		getLogger().debug("was cancelled.");
 	}
 
-	public final OperationState getState() {
+	public final synchronized OperationState getState() {
 		return state;
 	}
 
-	public final ByteBuffer getBuffer() {
+	public final synchronized ByteBuffer getBuffer() {
 		return cmd;
 	}
 
 	/**
 	 * Set the write buffer for this operation.
 	 */
-	protected final void setBuffer(ByteBuffer to) {
+	protected final synchronized void setBuffer(ByteBuffer to) {
 		assert to != null : "Trying to set buffer to null";
 		cmd=to;
 		cmd.mark();
@@ -100,7 +100,7 @@ protected final void setBuffer(ByteBuffer to) {
 	/**
 	 * Transition the state of this operation to the given state.
 	 */
-	protected final void transitionState(OperationState newState) {
+	protected final synchronized 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.
@@ -110,10 +110,6 @@ protected final void transitionState(OperationState newState) {
 		if(state == OperationState.COMPLETE) {
 			callback.complete();
 		}
-		if(state == OperationState.TIMEDOUT) {
-			cmd = null;
-			callback.complete();
-		}
 	}
 
 	public final void writing() {
@@ -160,26 +156,26 @@ public void setHandlingNode(MemcachedNode to) {
 	}
 
         @Override
-        public void timeOut() {
+        public synchronized void timeOut() {
 	    assert (state != OperationState.READING || state != OperationState.COMPLETE);
-	    this.transitionState(OperationState.TIMEDOUT);
             timedout = true;
+            callback.complete();
         }
 
         @Override
-        public boolean isTimedOut() {
+        public synchronized boolean isTimedOut() {
             return timedout;
         }
 
 	@Override
-	public boolean isTimedOut(long ttlMillis) {
+	public synchronized boolean isTimedOut(long ttlMillis) {
 		long elapsed = System.nanoTime();
 		long ttlNanos = ttlMillis * 1000 * 1000;
 		if (elapsed - creationTime > ttlNanos) {
 			assert (state != OperationState.READING || state != OperationState.COMPLETE);
-			this.transitionState(OperationState.TIMEDOUT);
 			timedOutUnsent = true;
 			timedout = true;
+			callback.complete();
 		} else {
 			// timedout would be false, but we cannot allow you to untimeout an operation
 			if (timedout) {

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

@@ -159,29 +159,31 @@ public final void fillWriteBuffer(boolean shouldOptimize) {
 			Operation o=getNextWritableOp();
 
 			while(o != null && toWrite < getWbuf().capacity()) {
-				assert o.getState() == OperationState.WRITING;
-
-				ByteBuffer obuf=o.getBuffer();
-				assert obuf != null : "Didn't get a write buffer from " + o;
-				int bytesToCopy=Math.min(getWbuf().remaining(),
-						obuf.remaining());
-				byte b[]=new byte[bytesToCopy];
-				obuf.get(b);
-				getWbuf().put(b);
-				getLogger().debug("After copying stuff from %s: %s",
-						o, getWbuf());
-				if(!o.getBuffer().hasRemaining()) {
-					o.writeComplete();
-					transitionWriteItem();
-
-					preparePending();
-					if(shouldOptimize) {
-						optimize();
+				synchronized(o) {
+					assert o.getState() == OperationState.WRITING;
+	
+					ByteBuffer obuf=o.getBuffer();
+					assert obuf != null : "Didn't get a write buffer from " + o;
+					int bytesToCopy=Math.min(getWbuf().remaining(),
+							obuf.remaining());
+					byte b[]=new byte[bytesToCopy];
+					obuf.get(b);
+					getWbuf().put(b);
+					getLogger().debug("After copying stuff from %s: %s",
+							o, getWbuf());
+					if(!o.getBuffer().hasRemaining()) {
+						o.writeComplete();
+						transitionWriteItem();
+	
+						preparePending();
+						if(shouldOptimize) {
+							optimize();
+						}
+	
+						o=getNextWritableOp();
 					}
-
-					o=getNextWritableOp();
+					toWrite += bytesToCopy;
 				}
-				toWrite += bytesToCopy;
 			}
 			getWbuf().flip();
 			assert toWrite <= getWbuf().capacity()
@@ -197,22 +199,24 @@ public final void fillWriteBuffer(boolean shouldOptimize) {
 	private Operation getNextWritableOp() {
 		Operation o = getCurrentWriteOp();
 		while (o != null && o.getState() == OperationState.WRITE_QUEUED) {
-			if (o.isCancelled()) {
-				getLogger().debug("Not writing cancelled op.");
-				Operation cancelledOp = removeCurrentWriteOp();
-				assert o == cancelledOp;
-			} else if (o.isTimedOut(defaultOpTimeout)) {
-				getLogger().debug("Not writing timed out op.");
-				Operation timedOutOp = removeCurrentWriteOp();
-				assert o == timedOutOp;
-			} else {
-				o.writing();
-				if (!(o instanceof TapAckOperationImpl)) {
-					readQ.add(o);
+			synchronized(o) {
+				if (o.isCancelled()) {
+					getLogger().debug("Not writing cancelled op.");
+					Operation cancelledOp = removeCurrentWriteOp();
+					assert o == cancelledOp;
+				} else if (o.isTimedOut(defaultOpTimeout)) {
+					getLogger().debug("Not writing timed out op.");
+					Operation timedOutOp = removeCurrentWriteOp();
+					assert o == timedOutOp;
+				} else {
+					o.writing();
+					if (!(o instanceof TapAckOperationImpl)) {
+						readQ.add(o);
+					}
+					return o;
 				}
-				return o;
+				o = getCurrentWriteOp();
 			}
-			o = getCurrentWriteOp();
 		}
 		return o;
 	}