CoralLog Throughput Test

CoralLog Throughput Test

package com.coralblocks.corallog.bench;

import java.io.File;
import java.nio.ByteBuffer;
import java.text.NumberFormat;
import java.util.Locale;

import com.coralblocks.coralbits.bench.Benchmarker;
import com.coralblocks.coralbits.util.SystemUtils;
import com.coralblocks.corallog.AsyncThread;
import com.coralblocks.corallog.Log;
import com.coralblocks.corallog.LogConfig;
import com.coralblocks.corallog.Logger;
import com.coralblocks.coralthreads.Affinity;

public class Throughput {
	
	public static void main(String[] args) throws Exception {
		
		NumberFormat nf = NumberFormat.getNumberInstance(Locale.US);
		
		int messagesPerBatch = Integer.parseInt(args[0]);
		int numberOfBatches = Integer.parseInt(args[1]);
		int msgSize = Integer.parseInt(args[2]);
		
		byte[] msgBytes = new byte[msgSize];
		
		// build a dummy message:
		for(int i = 0; i < msgBytes.length; i++) {
			msgBytes[i] = (byte) String.valueOf(i % 10).charAt(0);	
		}
		
		System.out.println();
		System.out.println("Messages per batch: " + nf.format(messagesPerBatch));
		System.out.println("Number of batches: " + numberOfBatches);
		System.out.println("Message size: " + msgSize + " bytes");
		
		System.out.println();
		
		int procToBindProducer = SystemUtils.getInt("procToBindProducer", -1);
		if (procToBindProducer != -1) Affinity.set(procToBindProducer);
		
		String filename = "throughput.log";
		
		LogConfig logConfig = new LogConfig(filename);
		logConfig.includeTimestamp = SystemUtils.getBoolean("includeTimestamp", false);
		
		Logger logger = Log.createLogger(logConfig);
		
		ByteBuffer bb = ByteBuffer.allocateDirect(1024);
		bb.put(msgBytes);
		bb.flip();
		
		int capacity = AsyncThread.getCapacity();
		boolean drainQueue = SystemUtils.getBoolean("drainQueue", true);

		int count = 0;
		
		while(count < numberOfBatches) {
			
			long start = System.nanoTime();
			
			for(int i = 0; i < messagesPerBatch; i++) {
				
				bb.position(0);
				
				logger.log(bb);
				
				if (drainQueue && (i + 1) % capacity == 0) {
					logger.drainAndWait();
				}
			}
			
			long time = System.nanoTime() - start;
			
			long latency = time / messagesPerBatch;
			double seconds = time / 1000000000.0;
			long ops = Math.round(messagesPerBatch / seconds);
			
			System.out.println("Batch " + (count + 1) + " took " + Benchmarker.convertNanoTime(time) + ": Latency = " + latency + " nanos per message | Throughput = " + nf.format(ops) + " messages/sec");
			
			count++;
		}
		
		System.out.println();
		
		logger.drainCloseAndWait();
		
		boolean deleteFile = SystemUtils.getBoolean("deleteFile", true);
		
		if (deleteFile) {
			File f = new File(filename);
			f.delete();
		}
		
		AsyncThread.drainAndDie(); // just so the vm will exit... (async thread is not daemon)
	}
}