// TODO: Mark it as deleted then loop to see if anybody is reading/writing to it
// If a write is after, keep the row by unmarking it as deleted (since the
// operation cancels)
// If a read is after, check the deleted flag, if its deleted, make sure to
// return key DNE
// Otherwise, return the key
// If at any point the queue is empty, and the row is marked as deleted, THEN
// remove it from the clientRequests
// Delete it from the cache as well!
// remove the top item from the list for this key
@Override
	public KVMessage putKV(final int clientPort, final String key, final String value)
			throws InvalidMessageException {
		logger.info(clientPort + "> PUT for key=" + key + " value=" + value);
		KVMessage res;
		StatusType putStat;
		// TODO: Cleanup the clientRequests after the requests are completed
		clientRequests.putIfAbsent(key, new ConcurrentNode(MAX_READS));
		NodeOperation op = value.equals("null") ? NodeOperation.DELETE : NodeOperation.WRITE;
		// add thread to back of list for this key - add is thread safe
		int[] node = { clientPort, op.getVal() };
		clientRequests.get(key).addToQueue(node);
		if (test) {
			logger.info(clientPort + "> !!!!===wait===!!!!");
			while (wait)
				;
			logger.info(clientPort + "> !!!!===DONE WAITING===!!!!");
		}
		// wait (spin) until threads turn and no one is reading
		// TODO: If a key gets deleted, I think clientRequests.get(key) would no longer
		// work
		while (clientRequests.get(key).peek()[0] != clientPort)
			;
		// while (clientRequests.get(key).peek()[0] != clientPort ||
		// clientRequests.get(key).availablePermits() != MAX_READS);
		logger.info(clientPort + "> !!!!===Finished spinning===!!!!");
		if (value.equals("null")) {
			// Delete the key
			logger.info(clientPort + "> Trying to delete record ...");
			putStat = StatusType.DELETE_SUCCESS;
			try {
				// TODO: Mark it as deleted then loop to see if anybody is reading/writing to it
				// If a write is after, keep the row by unmarking it as deleted (since the
				// operation cancels)
				// If a read is after, check the deleted flag, if its deleted, make sure to
				// return key DNE
				// Otherwise, return the key
				// If at any point the queue is empty, and the row is marked as deleted, THEN
				// remove it from the clientRequests
				// Delete it from the cache as well!
				// remove the top item from the list for this key
				if (!inStorage(key)) {
					logger.info(clientPort + "> Not in storage ...");
					putStat = StatusType.DELETE_ERROR;
				} else if (!clientRequests.get(key).isDeleted()) {
					logger.info(clientPort + "> Marked for deletion");
					clientRequests.get(key).setDeleted(true);
					Runnable pruneDelete = new Runnable() {
						@Override
						public void run() {
							logger.info(clientPort + "> Starting pruning thread");
							do {
								// logger.debug("Prune waiting");
							} while (!clientRequests.get(key).isEmpty());
							clientRequests.remove(key);
							if (cache != null) {
								cache.remove(key);
							}
							File file = new File(storageDirectory + key);
							file.delete();
							logger.info(clientPort + "> " + key + " successfully pruned");
						}
					};
					clientRequests.get(key).startPruning(pruneDelete);
				} else {
					logger.info(clientPort + "> Key does not exist - marked for deletion!");
					putStat = StatusType.DELETE_ERROR;
				}
			} catch (Exception e) {
				logger.info(clientPort + "> Deletion exception ...");
				logger.error(e);
				exceptionLogger(e);
				putStat = StatusType.DELETE_ERROR;
			}
		} else {
			// Insert/update the key
			logger.info(clientPort + "> Trying to insert/update record");
			try {
				if (!inStorage(key)) {
					// Inserting a new key
					logger.info(clientPort + "> Going to insert record");
					putStat = StatusType.PUT_SUCCESS;
				} else {
					// Updating a key
					logger.info(clientPort + "> Going to update record");
					putStat = StatusType.PUT_UPDATE;
					if (clientRequests.get(key).isDeleted()) {
						// Stop the deletion
						clientRequests.get(key).stopPruning();
						clientRequests.get(key).setDeleted(false);
					}
				}
				insertCache(key, value);
				FileWriter myWriter = new FileWriter(storageDirectory + key);
				myWriter.write(value);
				myWriter.close();
				// Sleep for a bit to let data take effect
				// Thread.sleep(1000);
			} catch (Exception e) {
				logger.error(e);
				exceptionLogger(e);
				putStat = StatusType.PUT_ERROR;
			}
		}
		// remove the top item from the list for this key
		removeTopQueue(key);
		logger.info(clientPort + "> !!!!===Removing from queue===!!!!");
		try {
			res = new KVMessage(key, value, putStat);
		} catch (InvalidMessageException ime) {
			throw ime;
		}
		return res;
	}
putStat = StatusType.DELETE_SUCCESS;
try {
// TODO: Mark it as deleted then loop to see if anybody is reading/writing to it
// If a write is after, keep the row by unmarking it as deleted (since the
// operation cancels)
// If a read is after, check the deleted flag, if its deleted, make sure to
// return key DNE
// Otherwise, return the key
// If at any point the queue is empty, and the row is marked as deleted, THEN
// remove it from the clientRequests
// Delete it from the cache as well!
// remove the top item from the list for this key
if (!inStorage(key)) {
logger.info(clientPort + "> Not in storage ...");
while (clientRequests.get(key).peek()[0] != clientPort)
;
// while (clientRequests.get(key).peek()[0] != clientPort ||
// clientRequests.get(key).availablePermits() != MAX_READS);
logger.info(clientPort + "> !!!!===Finished spinning===!!!!");
if (value.equals("null")) {
// Delete the key
logger.info(clientPort + "> Trying to delete record ...");
putStat = StatusType.DELETE_SUCCESS;
try {
// TODO: Mark it as deleted then loop to see if anybody is reading/writing to it
// If a write is after, keep the row by unmarking it as deleted (since the
// operation cancels)
// If a read is after, check the deleted flag, if its deleted, make sure to
// return key DNE
// Otherwise, return the key
// If at any point the queue is empty, and the row is marked as deleted, THEN
// remove it from the clientRequests
// Delete it from the cache as well!
// remove the top item from the list for this key
if (!inStorage(key)) {
logger.info(clientPort + "> Not in storage ...");
putStat = StatusType.DELETE_ERROR;
} else if (!clientRequests.get(key).isDeleted()) {
logger.info(clientPort + "> Marked for deletion");
clientRequests.get(key).setDeleted(true);
Runnable pruneDelete = new Runnable() {
@Override
public void run() {
logger.info(clientPort + "> Starting pruning thread");
clientRequests.get(key).addToQueue(node);
if (test) {
logger.info(clientPort + "> !!!!===wait===!!!!");
while (wait)
;
logger.info(clientPort + "> !!!!===DONE WAITING===!!!!");
}
// wait (spin) until threads turn and no one is reading
// TODO: If a key gets deleted, I think clientRequests.get(key) would no longer
// work
while (clientRequests.get(key).peek()[0] != clientPort)
;
// while (clientRequests.get(key).peek()[0] != clientPort ||
// clientRequests.get(key).availablePermits() != MAX_READS);
logger.info(clientPort + "> !!!!===Finished spinning===!!!!");
if (value.equals("null")) {
// Delete the key
logger.info(clientPort + "> Trying to delete record ...");
putStat = StatusType.DELETE_SUCCESS;
try {
// TODO: Mark it as deleted then loop to see if anybody is reading/writing to it
// If a write is after, keep the row by unmarking it as deleted (since the
// operation cancels)
// If a read is after, check the deleted flag, if its deleted, make sure to
// return key DNE
// Otherwise, return the key
// If at any point the queue is empty, and the row is marked as deleted, THEN
// remove it from the clientRequests
// Delete it from the cache as well!
// remove the top item from the list for this key
if (!inStorage(key)) {
logger.info(clientPort + "> Not in storage ...");
putStat = StatusType.DELETE_ERROR;
} else if (!clientRequests.get(key).isDeleted()) {
logger.info(clientPort + "> Marked for deletion");
clientRequests.get(key).setDeleted(true);
Runnable pruneDelete = new Runnable() {
@Override
public void run() {
logger.info(clientPort + "> Starting pruning thread");
do {
// logger.debug("Prune waiting");
} while (!clientRequests.get(key).isEmpty());
clientRequests.remove(key);
if (cache != null) {
cache.remove(key);
}
File file = new File(storageDirectory + key);
file.delete();
logger.info(clientPort + "> " + key + " successfully pruned");