Improved performance of write queue processing during timeouts

There were two major issues here. First if there are a lot of
timeouts in a row then we processing the write queue we removed
them only one at a time. We now are able to process the entire
write queue each time the handleIO function is called. The
second issue was that it is possible to send timed out requests
to the server. This would happen if there was one not timed out
request followed by some timed out requests. Since in the while
loop we don't check to see if an operation is timed out (or
cancelled) we would send it anyways. This commit addresses both
of these issues by adding a getNextWritable() op which only
operations that should be written. If the write queue is empty
it returns null.

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

Author: Mike Wiederhold

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

@@ -156,26 +156,8 @@ private boolean preparePending() {
 	public final void fillWriteBuffer(boolean shouldOptimize) {
 		if(toWrite == 0 && readQ.remainingCapacity() > 0) {
 			getWbuf().clear();
-			Operation o=getCurrentWriteOp();
+			Operation o=getNextWritableOp();
 
-			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();
-					if (!(o instanceof TapAckOperationImpl)) {
-						readQ.add(o);
-					}
-				}
-			}
 			while(o != null && toWrite < getWbuf().capacity()) {
 				assert o.getState() == OperationState.WRITING;
 
@@ -197,13 +179,7 @@ public final void fillWriteBuffer(boolean shouldOptimize) {
 						optimize();
 					}
 
-					o=getCurrentWriteOp();
-					if (o != null) {
-						o.writing();
-						if (!(o instanceof TapAckOperationImpl)) {
-							readQ.add(o);
-						}
-					}
+					o=getNextWritableOp();
 				}
 				toWrite += bytesToCopy;
 			}
@@ -218,6 +194,29 @@ 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);
+				}
+				return o;
+			}
+			o = getCurrentWriteOp();
+		}
+		return o;
+	}
+
 	/* (non-Javadoc)
 	 * @see net.spy.memcached.MemcachedNode#transitionWriteItem()
 	 */