REZKTestClientReadWriteConflict.java

package refit.application.rezk;

import java.util.Random;

import refit.application.rezk.common.REZKStat;
import refit.client.REFITClientLibrary;
import refit.config.REFITConfig;
import refit.util.REFITTime;

public class REZKTestClientReadWriteConflict extends REZKTestClient {
	private static final String[] paths = new String[REFITConfig.ACTIVE_NODES];

	private Random rand;

	public REZKTestClientReadWriteConflict(REFITClientLibrary service) {
		super(service);

		synchronized (paths) {
			if (paths[0] == null) {
				for (int i = 0; i < paths.length; i++) {
					paths[i] = String.format("%s%05d", "/node", i);
				}
			}
		}

		rand = new Random(service.getNodeID());
	}

	@Override
	protected void work() throws Exception {
		if (rand.nextFloat() < REFITConfig.READ_WRITE_RATIO) {
			read();
		} else {
			write();
		}
	}

	private void read() throws Exception {
		// Invoke operation
		long startTime = REFITTime.nanoTime.getAsLong();
		int idx = rand.nextInt(paths.length);
		byte[] data = client.getData(paths[idx], false, null, REFITConfig.ONLY_CONSISTENT);
		long endTime = REFITTime.nanoTime.getAsLong();
		service.getStatistics().event((long) ((endTime - startTime) / 1000f), service.getNodeID(), service.isReadWriteConflict() ? "rw" : "r");

//		if (!REZKTestServer.verifyNodeData(idx, data)) throw new Exception("Node contains invalid data");
	}


	private void write() throws Exception {
		byte[] data = REZKTestServer.createNodeData(rand.nextInt(1000));
		// Invoke operation
		long startTime = REFITTime.nanoTime.getAsLong();
		int idx = rand.nextInt(paths.length);
		REZKStat stat = client.setData(paths[idx], data, -1);
		long endTime = REFITTime.nanoTime.getAsLong();
		service.getStatistics().event((long) ((endTime - startTime) / 1000f), service.getNodeID(), "w");

		// very unlikely
		if (stat.dataLength != data.length) throw new Exception("Node data size mismatch");
	}
}