<commit_before>/*
 * arcus-java-client : Arcus Java client
 * Copyright 2010-2014 NAVER Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.spy.memcached.protocol;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.StringTokenizer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import net.spy.memcached.ArcusReplNodeAddress;
import net.spy.memcached.CacheManager;
import net.spy.memcached.MemcachedNode;
import net.spy.memcached.MemcachedReplicaGroup;
import net.spy.memcached.compat.SpyObject;
import net.spy.memcached.ops.APIType;
import net.spy.memcached.ops.Operation;
import net.spy.memcached.ops.OperationState;

/**
 * Represents a node with the memcached cluster, along with buffering and
 * operation queues.
 */
public abstract class TCPMemcachedNodeImpl extends SpyObject
	implements MemcachedNode {

	private final SocketAddress socketAddress;
	private final ByteBuffer rbuf;
	private final ByteBuffer wbuf;
	protected final BlockingQueue<Operation> writeQ;
	private final BlockingQueue<Operation> readQ;
	private final BlockingQueue<Operation> inputQueue;
	private final long opQueueMaxBlockTime;
	// This has been declared volatile so it can be used as an availability
	// indicator.
	private volatile int reconnectAttempt=1;
	private SocketChannel channel;
	private int toWrite=0;
	protected Operation optimizedOp=null;
	private volatile SelectionKey sk=null;
	private boolean shouldAuth=false;
	private CountDownLatch authLatch;
	private ArrayList<Operation> reconnectBlocked;
	private String version=null;
	private boolean enabledMGetOp=false;

	// operation Future.get timeout counter
	private final AtomicInteger continuousTimeout = new AtomicInteger(0);
	private boolean toRatioEnabled = false;
	private int[] toCountArray;
	private final static int MAX_TOCOUNT = 100;   /* to count array size */
	private int toCountIdx;         /* to count array index */
	private int toRatioMax;         /* maximum timeout ratio */
	private int toRatioNow;         /* current timeout ratio */
	private Lock toRatioLock = new ReentrantLock();

	/* # of operations added into inputQueue as a hint.
	 * If we need a correct count, AtomicLong object must be used.
	 */
	private volatile long addOpCount;

	// fake node
	private boolean isFake = false; 
	
	/* ENABLE_REPLICATION if */
	private MemcachedReplicaGroup replicaGroup;

	/* ENABLE_REPLICATION end */
	public boolean isFake() {
		return isFake;
	}

	private void resetTimeoutRatioCount() {
		if (toRatioEnabled) {
			toRatioLock.lock();
			for (int i=0; i < MAX_TOCOUNT; i++) {
				toCountArray[i] = 0;
			}
			toCountIdx = -1;
			toRatioMax = 0;
			toRatioNow = 0;
			toRatioLock.unlock();
		}
	}

	private void addTimeoutRatioCount(boolean timedOut) {
		if (toRatioEnabled) {
			toRatioLock.lock();
			if ((++toCountIdx) >= MAX_TOCOUNT)
				toCountIdx = 0;
			if (toCountArray[toCountIdx] > 0) {
				toRatioNow -= toCountArray[toCountIdx];
				toCountArray[toCountIdx] = 0;
			}
			if (timedOut) {
				toCountArray[toCountIdx] = 1;
				toRatioNow += 1;
				if (toRatioNow > toRatioMax)
					toRatioMax = toRatioNow;
			}
			toRatioLock.unlock();
		}
	}

	public TCPMemcachedNodeImpl(SocketAddress sa, SocketChannel c,
			int bufSize, BlockingQueue<Operation> rq,
			BlockingQueue<Operation> wq, BlockingQueue<Operation> iq,
			long opQueueMaxBlockTime, boolean waitForAuth) {
		super();
		assert sa != null : "No SocketAddress";
		assert c != null : "No SocketChannel";
		assert bufSize > 0 : "Invalid buffer size: " + bufSize;
		assert rq != null : "No operation read queue";
		assert wq != null : "No operation write queue";
		assert iq != null : "No input queue";
		/* ENABLE_REPLICATION if */
		if (sa instanceof ArcusReplNodeAddress) {
			socketAddress = new ArcusReplNodeAddress((ArcusReplNodeAddress) sa);
		} else {
			socketAddress = sa;
		}
		/* ENABLE_REPLICATION else */
		/*
		socketAddress=sa;
		*/
		/* ENABLE_REPLICATION end */
		setChannel(c);
		rbuf=ByteBuffer.allocate(bufSize);
		wbuf=ByteBuffer.allocate(bufSize);
		getWbuf().clear();
		readQ=rq;
		writeQ=wq;
		inputQueue=iq;
		addOpCount=0;
		this.opQueueMaxBlockTime = opQueueMaxBlockTime;
		shouldAuth = waitForAuth;
		setupForAuth("init authentication");

		// is this a fake node?
		if (sa instanceof InetSocketAddress) {
			InetSocketAddress inetSockAddr = (InetSocketAddress)sa;
			String ipport = inetSockAddr.getAddress() + ":" + inetSockAddr.getPort();
			isFake = CacheManager.FAKE_SERVER_NODE.equals(ipport);
		}
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#copyInputQueue()
	 */
	public final void copyInputQueue() {
		Collection<Operation> tmp=new ArrayList<Operation>();

		// don't drain more than we have space to place
		inputQueue.drainTo(tmp, writeQ.remainingCapacity());

		writeQ.addAll(tmp);
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#destroyInputQueue()
	 */
	public Collection<Operation> destroyInputQueue() {
		Collection<Operation> rv=new ArrayList<Operation>();
		inputQueue.drainTo(rv);
		return rv;
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#destroyWriteQueue()
	 */
	public Collection<Operation> destroyWriteQueue(boolean resend) {
		Collection<Operation> rv=new ArrayList<Operation>();
		writeQ.drainTo(rv);
		if (resend) {
			for (Operation o : rv) {
				if (o.getState() == OperationState.WRITING && o.getBuffer() != null) {
					o.getBuffer().reset(); // buffer offset reset
				} else {
					o.initialize(); // write completed or not yet initialized
				}
			}
		}

		return rv;
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#setupResend()
	 */
	public final void setupResend(boolean cancelWrite, String cause) {
		// First, reset the current write op, or cancel it if we should
		// be authenticating
		Operation op=getCurrentWriteOp();
		if((cancelWrite || shouldAuth) && op != null) {
		    op.cancel(cause);
		} else if(op != null) {
			ByteBuffer buf=op.getBuffer();
			if(buf != null) {
				buf.reset();
			} else {
				getLogger().info("No buffer for current write op, removing");
				removeCurrentWriteOp();
			}
		}
		// Now cancel all the pending read operations.  Might be better to
		// to requeue them.
		while(hasReadOp()) {
			op=removeCurrentReadOp();
			if (op != getCurrentWriteOp()) {
				getLogger().warn("Discarding partially completed op: %s", op);
				op.cancel(cause);
			}
		}

		while((cancelWrite || shouldAuth) && hasWriteOp()) {
			op=removeCurrentWriteOp();
			getLogger().warn("Discarding partially completed op: %s", op);
			op.cancel(cause);
		}


		getWbuf().clear();
		getRbuf().clear();
		toWrite=0;
	}

	// Prepare the pending operations.  Return true if there are any pending
	// ops
	private boolean preparePending() {
		// Copy the input queue into the write queue.
		copyInputQueue();

		// Now check the ops
		Operation nextOp=getCurrentWriteOp();
		while(nextOp != null && nextOp.isCancelled()) {
			getLogger().info("Removing cancelled operation: %s", nextOp);
			removeCurrentWriteOp();
			nextOp=getCurrentWriteOp();
		}
		return nextOp != null;
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#fillWriteBuffer(boolean)
	 */
	public final void fillWriteBuffer(boolean shouldOptimize) {
		if(toWrite == 0 && readQ.remainingCapacity() > 0) {
			getWbuf().clear();
			Operation o=getCurrentWriteOp();
			while(o != null && toWrite < getWbuf().capacity()) {
				assert o.getState() == OperationState.WRITING;
				// This isn't the most optimal way to do this, but it hints
				// at a larger design problem that may need to be taken care
				// if in the bowels of the client.
				// In practice, readQ should be small, however.
				if(!readQ.contains(o)) {
					readQ.add(o);
				}

				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) {
						/* FIXME
						 * MGetOperation needs optimization.
						 * Temporarily do not optimize on MGetOperation.
						 */
						if (writeQ.peek().getAPIType() != APIType.MGET) {
							optimize();
						}
					}

					o=getCurrentWriteOp();
				}
				toWrite += bytesToCopy;
			}
			getWbuf().flip();
			assert toWrite <= getWbuf().capacity()
				: "toWrite exceeded capacity: " + this;
			assert toWrite == getWbuf().remaining()
				: "Expected " + toWrite + " remaining, got "
				+ getWbuf().remaining();
		} else {
			getLogger().debug("Buffer is full, skipping");
		}
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#transitionWriteItem()
	 */
	public final void transitionWriteItem() {<commit_msg>FIX: WriteQueue check for peek Operation<commit_after>/*
 * arcus-java-client : Arcus Java client
 * Copyright 2010-2014 NAVER Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.spy.memcached.protocol;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.StringTokenizer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import net.spy.memcached.ArcusReplNodeAddress;
import net.spy.memcached.CacheManager;
import net.spy.memcached.MemcachedNode;
import net.spy.memcached.MemcachedReplicaGroup;
import net.spy.memcached.compat.SpyObject;
import net.spy.memcached.ops.APIType;
import net.spy.memcached.ops.Operation;
import net.spy.memcached.ops.OperationState;

/**
 * Represents a node with the memcached cluster, along with buffering and
 * operation queues.
 */
public abstract class TCPMemcachedNodeImpl extends SpyObject
	implements MemcachedNode {

	private final SocketAddress socketAddress;
	private final ByteBuffer rbuf;
	private final ByteBuffer wbuf;
	protected final BlockingQueue<Operation> writeQ;
	private final BlockingQueue<Operation> readQ;
	private final BlockingQueue<Operation> inputQueue;
	private final long opQueueMaxBlockTime;
	// This has been declared volatile so it can be used as an availability
	// indicator.
	private volatile int reconnectAttempt=1;
	private SocketChannel channel;
	private int toWrite=0;
	protected Operation optimizedOp=null;
	private volatile SelectionKey sk=null;
	private boolean shouldAuth=false;
	private CountDownLatch authLatch;
	private ArrayList<Operation> reconnectBlocked;
	private String version=null;
	private boolean enabledMGetOp=false;

	// operation Future.get timeout counter
	private final AtomicInteger continuousTimeout = new AtomicInteger(0);
	private boolean toRatioEnabled = false;
	private int[] toCountArray;
	private final static int MAX_TOCOUNT = 100;   /* to count array size */
	private int toCountIdx;         /* to count array index */
	private int toRatioMax;         /* maximum timeout ratio */
	private int toRatioNow;         /* current timeout ratio */
	private Lock toRatioLock = new ReentrantLock();

	/* # of operations added into inputQueue as a hint.
	 * If we need a correct count, AtomicLong object must be used.
	 */
	private volatile long addOpCount;

	// fake node
	private boolean isFake = false; 
	
	/* ENABLE_REPLICATION if */
	private MemcachedReplicaGroup replicaGroup;

	/* ENABLE_REPLICATION end */
	public boolean isFake() {
		return isFake;
	}

	private void resetTimeoutRatioCount() {
		if (toRatioEnabled) {
			toRatioLock.lock();
			for (int i=0; i < MAX_TOCOUNT; i++) {
				toCountArray[i] = 0;
			}
			toCountIdx = -1;
			toRatioMax = 0;
			toRatioNow = 0;
			toRatioLock.unlock();
		}
	}

	private void addTimeoutRatioCount(boolean timedOut) {
		if (toRatioEnabled) {
			toRatioLock.lock();
			if ((++toCountIdx) >= MAX_TOCOUNT)
				toCountIdx = 0;
			if (toCountArray[toCountIdx] > 0) {
				toRatioNow -= toCountArray[toCountIdx];
				toCountArray[toCountIdx] = 0;
			}
			if (timedOut) {
				toCountArray[toCountIdx] = 1;
				toRatioNow += 1;
				if (toRatioNow > toRatioMax)
					toRatioMax = toRatioNow;
			}
			toRatioLock.unlock();
		}
	}

	public TCPMemcachedNodeImpl(SocketAddress sa, SocketChannel c,
			int bufSize, BlockingQueue<Operation> rq,
			BlockingQueue<Operation> wq, BlockingQueue<Operation> iq,
			long opQueueMaxBlockTime, boolean waitForAuth) {
		super();
		assert sa != null : "No SocketAddress";
		assert c != null : "No SocketChannel";
		assert bufSize > 0 : "Invalid buffer size: " + bufSize;
		assert rq != null : "No operation read queue";
		assert wq != null : "No operation write queue";
		assert iq != null : "No input queue";
		/* ENABLE_REPLICATION if */
		if (sa instanceof ArcusReplNodeAddress) {
			socketAddress = new ArcusReplNodeAddress((ArcusReplNodeAddress) sa);
		} else {
			socketAddress = sa;
		}
		/* ENABLE_REPLICATION else */
		/*
		socketAddress=sa;
		*/
		/* ENABLE_REPLICATION end */
		setChannel(c);
		rbuf=ByteBuffer.allocate(bufSize);
		wbuf=ByteBuffer.allocate(bufSize);
		getWbuf().clear();
		readQ=rq;
		writeQ=wq;
		inputQueue=iq;
		addOpCount=0;
		this.opQueueMaxBlockTime = opQueueMaxBlockTime;
		shouldAuth = waitForAuth;
		setupForAuth("init authentication");

		// is this a fake node?
		if (sa instanceof InetSocketAddress) {
			InetSocketAddress inetSockAddr = (InetSocketAddress)sa;
			String ipport = inetSockAddr.getAddress() + ":" + inetSockAddr.getPort();
			isFake = CacheManager.FAKE_SERVER_NODE.equals(ipport);
		}
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#copyInputQueue()
	 */
	public final void copyInputQueue() {
		Collection<Operation> tmp=new ArrayList<Operation>();

		// don't drain more than we have space to place
		inputQueue.drainTo(tmp, writeQ.remainingCapacity());

		writeQ.addAll(tmp);
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#destroyInputQueue()
	 */
	public Collection<Operation> destroyInputQueue() {
		Collection<Operation> rv=new ArrayList<Operation>();
		inputQueue.drainTo(rv);
		return rv;
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#destroyWriteQueue()
	 */
	public Collection<Operation> destroyWriteQueue(boolean resend) {
		Collection<Operation> rv=new ArrayList<Operation>();
		writeQ.drainTo(rv);
		if (resend) {
			for (Operation o : rv) {
				if (o.getState() == OperationState.WRITING && o.getBuffer() != null) {
					o.getBuffer().reset(); // buffer offset reset
				} else {
					o.initialize(); // write completed or not yet initialized
				}
			}
		}

		return rv;
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#setupResend()
	 */
	public final void setupResend(boolean cancelWrite, String cause) {
		// First, reset the current write op, or cancel it if we should
		// be authenticating
		Operation op=getCurrentWriteOp();
		if((cancelWrite || shouldAuth) && op != null) {
		    op.cancel(cause);
		} else if(op != null) {
			ByteBuffer buf=op.getBuffer();
			if(buf != null) {
				buf.reset();
			} else {
				getLogger().info("No buffer for current write op, removing");
				removeCurrentWriteOp();
			}
		}
		// Now cancel all the pending read operations.  Might be better to
		// to requeue them.
		while(hasReadOp()) {
			op=removeCurrentReadOp();
			if (op != getCurrentWriteOp()) {
				getLogger().warn("Discarding partially completed op: %s", op);
				op.cancel(cause);
			}
		}

		while((cancelWrite || shouldAuth) && hasWriteOp()) {
			op=removeCurrentWriteOp();
			getLogger().warn("Discarding partially completed op: %s", op);
			op.cancel(cause);
		}


		getWbuf().clear();
		getRbuf().clear();
		toWrite=0;
	}

	// Prepare the pending operations.  Return true if there are any pending
	// ops
	private boolean preparePending() {
		// Copy the input queue into the write queue.
		copyInputQueue();

		// Now check the ops
		Operation nextOp=getCurrentWriteOp();
		while(nextOp != null && nextOp.isCancelled()) {
			getLogger().info("Removing cancelled operation: %s", nextOp);
			removeCurrentWriteOp();
			nextOp=getCurrentWriteOp();
		}
		return nextOp != null;
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#fillWriteBuffer(boolean)
	 */
	public final void fillWriteBuffer(boolean shouldOptimize) {
		if(toWrite == 0 && readQ.remainingCapacity() > 0) {
			getWbuf().clear();
			Operation o=getCurrentWriteOp();
			while(o != null && toWrite < getWbuf().capacity()) {
				assert o.getState() == OperationState.WRITING;
				// This isn't the most optimal way to do this, but it hints
				// at a larger design problem that may need to be taken care
				// if in the bowels of the client.
				// In practice, readQ should be small, however.
				if(!readQ.contains(o)) {
					readQ.add(o);
				}

				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) {
						/* FIXME
						 * MGetOperation needs optimization.
						 * Temporarily do not optimize on MGetOperation.
						 */
						if (writeQ.peek() != null && writeQ.peek().getAPIType() != APIType.MGET) {
							optimize();
						}
					}

					o=getCurrentWriteOp();
				}
				toWrite += bytesToCopy;
			}
			getWbuf().flip();
			assert toWrite <= getWbuf().capacity()
				: "toWrite exceeded capacity: " + this;
			assert toWrite == getWbuf().remaining()
				: "Expected " + toWrite + " remaining, got "
				+ getWbuf().remaining();
		} else {
			getLogger().debug("Buffer is full, skipping");
		}
	}

	/* (non-Javadoc)
	 * @see net.spy.memcached.MemcachedNode#transitionWriteItem()
	 */
	public final void transitionWriteItem() {
{'fraction_non_alphanumeric': 0.11695906432748537, 'fraction_numerical': 0.003327283726557774, 'mean_word_length': 3.8489366902957713, 'pattern_counts': {'":': 2, '<': 39, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6000342', 'n_tokens_mistral': 7042, 'n_tokens_neox': 6156, 'n_words': 1651}