Second try: large table mods merged against current treilly master

.gitignore

@@ -1 +1,6 @@
 /elasticsearch/target/
+.project
+.classpath
+*/target/*
+.settings
+*~

core/src/main/java/com/yahoo/ycsb/Client.java

@@ -25,12 +25,16 @@
 import com.yahoo.ycsb.measurements.Measurements;
 import com.yahoo.ycsb.measurements.exporter.MeasurementsExporter;
 import com.yahoo.ycsb.measurements.exporter.TextMeasurementsExporter;
+import com.yahoo.ycsb.statusreporter.StatusReporter;
 
 //import org.apache.log4j.BasicConfigurator;
 
 /**
  * A thread to periodically show the status of the experiment, to reassure you that progress is being made.
- * 
+ * Launched only if the "status" flag (-s) is specified on the command line.
+ * YCSB param "statusinterval" can be used to override the default status reporting interval of 10000 milliseconds.
+ * By default status messages are printed to stdout; YCSB param "statusreporter" specifies an optional plugin to 
+ * route status information to some additional destination.
  * @author cooperb
  *
  */
@@ -39,17 +43,40 @@ class StatusThread extends Thread
 	Vector<Thread> _threads;
 	String _label;
 	boolean _standardstatus;
+	long _reportInterval;
+	StatusReporter _reporter;
 
 	/**
 	 * The interval for reporting status.
 	 */
 	public static final long sleeptime=10000;
 
-	public StatusThread(Vector<Thread> threads, String label, boolean standardstatus)
+	public StatusThread(Vector<Thread> threads, String label, boolean standardstatus, Properties props)
 	{
 		_threads=threads;
 		_label=label;
 		_standardstatus=standardstatus;
+		if (props.containsKey("statusinterval"))
+		{
+		    _reportInterval=Long.parseLong(props.getProperty("statusinterval"));
+		}
+		else
+		    _reportInterval=sleeptime;
+		if (props.containsKey("statusreporter"))
+		{
+		    String reporterStr=props.getProperty("statusreporter");
+          try
+            {
+              _reporter = (StatusReporter) Class.forName(reporterStr).getConstructor().newInstance();
+              _reporter.configure(props);
+            } catch (Exception e)
+            {
+                System.err.println("Could not find status reporter " + reporterStr + ", ignoring custom statusreporter.");
+                _reporter = null;
+            }
+		}
+		else
+		    _reporter=null;
 	}
 
 	/**
@@ -68,7 +95,7 @@ public void run()
 		{
 			alldone=true;
 
-			int totalops=0;
+			long totalops=0;
 
 			//terminate this thread when all the worker threads are done
 			for (Thread t : _threads)
@@ -94,34 +121,44 @@ public void run()
 
 			DecimalFormat d = new DecimalFormat("#.##");
 
+			// getSummary() clears counters, so it can be called only once per loop
+			String measurementSummary=Measurements.getMeasurements().getSummary();
+
 			if (totalops==0)
 			{
-				System.err.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+Measurements.getMeasurements().getSummary());
+				System.err.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+measurementSummary);
 			}
 			else
 			{
-				System.err.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+d.format(curthroughput)+" current ops/sec; "+Measurements.getMeasurements().getSummary());
+				System.err.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+d.format(curthroughput)+" current ops/sec; "+measurementSummary);
 			}
 
 			if (_standardstatus)
 			{
 			if (totalops==0)
 			{
-				System.out.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+Measurements.getMeasurements().getSummary());
+				System.out.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+measurementSummary);
 			}
 			else
 			{
-				System.out.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+d.format(curthroughput)+" current ops/sec; "+Measurements.getMeasurements().getSummary());
+				System.out.println(_label+" "+(interval/1000)+" sec: "+totalops+" operations; "+d.format(curthroughput)+" current ops/sec; "+ measurementSummary);
 			}
 			}
 
+			if (_reporter != null)
+			{
+			    _reporter.report(interval, totalops, curthroughput, measurementSummary);
+			}
 			try
 			{
-				sleep(sleeptime);
+				sleep(_reportInterval);
 			}
 			catch (InterruptedException e)
 			{
-				//do nothing
+				if (_reporter != null)
+				{
+				    _reporter.close();
+				}
 			}
 
 		}
@@ -140,10 +177,10 @@ class ClientThread extends Thread
 	DB _db;
 	boolean _dotransactions;
 	Workload _workload;
-	int _opcount;
+	long _opcount;
 	double _target;
 
-	int _opsdone;
+	long _opsdone;
 	int _threadid;
 	int _threadcount;
 	Object _workloadstate;
@@ -162,7 +199,7 @@ class ClientThread extends Thread
 	 * @param opcount the number of operations (transactions or inserts) to do
 	 * @param targetperthreadperms target number of operations per thread per ms
 	 */
-	public ClientThread(DB db, boolean dotransactions, Workload workload, int threadid, int threadcount, Properties props, int opcount, double targetperthreadperms)
+	public ClientThread(DB db, boolean dotransactions, Workload workload, int threadid, int threadcount, Properties props, long opcount, double targetperthreadperms)
 	{
 		//TODO: consider removing threadcount and threadid
 		_db=db;
@@ -177,7 +214,7 @@ public ClientThread(DB db, boolean dotransactions, Workload workload, int thread
 		//System.out.println("Interval = "+interval);
 	}
 
-	public int getOpsDone()
+	public long getOpsDone()
 	{
 		return _opsdone;
 	}
@@ -384,7 +421,7 @@ public static boolean checkRequiredProperties(Properties props)
 	 * loaded from conf.
 	 * @throws IOException Either failed to write to output stream or failed to close it.
 	 */
-	private static void exportMeasurements(Properties props, int opcount, long runtime)
+	private static void exportMeasurements(Properties props, long opcount, long runtime)
 			throws IOException
 	{
 		MeasurementsExporter exporter = null;
@@ -684,22 +721,22 @@ public void run()
 
 		//run the workload
 
-		System.err.println("Starting test.");
+		System.err.println("Starting test...");
 
-		int opcount;
+		long opcount;
 		if (dotransactions)
 		{
-			opcount=Integer.parseInt(props.getProperty(OPERATION_COUNT_PROPERTY,"0"));
+			opcount=Long.parseLong(props.getProperty(OPERATION_COUNT_PROPERTY,"0"));
 		}
 		else
 		{
 			if (props.containsKey(INSERT_COUNT_PROPERTY))
 			{
-				opcount=Integer.parseInt(props.getProperty(INSERT_COUNT_PROPERTY,"0"));
+				opcount=Long.parseLong(props.getProperty(INSERT_COUNT_PROPERTY,"0"));
 			}
 			else
 			{
-				opcount=Integer.parseInt(props.getProperty(RECORD_COUNT_PROPERTY,"0"));
+				opcount=Long.parseLong(props.getProperty(RECORD_COUNT_PROPERTY,"0"));
 			}
 		}
 
@@ -733,7 +770,7 @@ public void run()
 			{
 				standardstatus=true;
 			}	
-			statusthread=new StatusThread(threads,label,standardstatus);
+			statusthread=new StatusThread(threads,label,standardstatus,props);
 			statusthread.start();
 		}
 
@@ -751,7 +788,7 @@ public void run()
       terminator.start();
     }
 
-    int opsDone = 0;
+    long opsDone = 0;
 
 		for (Thread t : threads)
 		{

core/src/main/java/com/yahoo/ycsb/generator/FocusedZipfianGenerator.java

@@ -0,0 +1,77 @@
+package com.yahoo.ycsb.generator;
+
+/**
+ * An extension of the zipfian genarator that focuses the hotspots on the center of fractions of the
+ * keyspace.   Centering is acheived by adding an centered offset to the zipfian value and treating odd/even 
+ * as +/- the center.   Focusing increases as the number of clients is added to keep the percentages of requests
+ * hitting a particular slice constant, this is achieved by using an array of ZipfianGenerators for each power
+ * of two leading to the number of clients.
+ */
+public class FocusedZipfianGenerator extends LongGenerator 
+{
+	ZipfianGenerator gen;
+	int genIndex;
+	long _itemcount,_offset;
+
+	/**
+	 * 
+	 * @param max
+	 * @param num numerator, or client id should be 0 -> denom
+	 * @param denom denominator, or number of clients
+	 */
+	public FocusedZipfianGenerator(long itemcount, long num, long denom)
+	{
+		_itemcount=itemcount;
+		_offset = (_itemcount / (2*denom)) * (num*2+1);
+		gen = new ZipfianGenerator(0,_itemcount);
+	}	
+
+	/**
+	 * Create a focused distribution that limits it self to a fraction of the keyspace instead of hitting all keys
+	 * @param itemcount size of entire key space
+	 * @param num numerator or client id, tells us what slice to hit
+	 * @param denom denominator or number of clients
+	 * @param partition dummy flag to separate different constructors
+	 */
+	public FocusedZipfianGenerator(long itemcount, long num, long denom, boolean partition)
+	{
+		_itemcount = itemcount / denom;
+		_offset = (itemcount / (2*denom)) * (num*2+1);
+		gen = new ZipfianGenerator(0, _itemcount);
+	}
+
+	/**************************************************************************************************/
+
+	/**
+	 * Return the next int in the sequence (kept for backwards compatibility).
+	 */
+	public int nextInt() {
+		return (int)nextLong();
+	}
+
+	/**
+	 * Return the next long in the sequence.
+	 */
+	@Override
+	public long nextLong()
+	{
+		long ret=gen.nextLong();
+
+		// odd numbers are greater than center, even less
+		if (ret % 2 == 0)
+			ret = -1 * ret;
+
+		// since we spread odd/even up/down we must divide by two to avoid holes.
+		ret /= 2;
+
+		ret=_offset + ret;
+		setLastLong(ret);
+		return ret;
+	}
+
+	@Override
+	public double mean() 
+	{
+		return 0;
+	}
+}

core/src/main/java/com/yahoo/ycsb/statusreporter/HttpStatusReporter.java

@@ -0,0 +1,91 @@
+package com.yahoo.ycsb.statusreporter;
+
+//import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.Properties;
+import java.util.regex.Pattern;
+
+
+/**
+ * Direct status information to a remote HTTP listener.
+ * Enable this plugin by defining YCSB parameter statusreporter=com.yahoo.ycsb.statusreporter.HttpStatusReporter
+ * YCSB param statusreporter.httpurl defines the endpoint to which the status message will be posted; if that URL
+ * includes the string "%c", that string gets replaced with the value of the "clientid" param.
+ * The output HTTP message has no body, but includes a list of http parameters identifying the clientid, interval
+ * and statistics for the current test interval.  
+ * @author dhentchel
+ */
+public class HttpStatusReporter implements StatusReporter {
+    String _clientid;
+    String _httpUrl;
+
+    public boolean configure(Properties props) {
+        _clientid = props.getProperty("clientid");
+        _httpUrl = props.getProperty("statusreporter.httpurl").replaceAll("%c", _clientid);
+        return true;
+    }
+
+    /**
+     * Post a list of HTTP parameters to the configured URL.
+     * Passes statistics as URL parameters, i.e. <_httpUrl>?parm1=value&parm2=value...
+     * Where parms always include client, interval, totalops and tps;
+     * and may include latency measures for read, update, insert, delete or scan.
+     */
+    public void report(long interval, long totalops, double curthroughput, String measures) {
+        String urlParameters = "client=" + _clientid + "&interval=" + interval + "&totalops=" + totalops + "&tps=" + curthroughput;
+
+       String[] parts = Pattern.compile("]|\\[").split(measures.trim());
+        for (String m : parts)
+        {
+            if (m.contains("="))
+            {
+                String[] vals = Pattern.compile("\\s|=").split(m.trim());
+                if (vals.length == 3)
+                {
+                    urlParameters = urlParameters + "&" + vals[0].toLowerCase() + "=" + vals[2];
+                }
+            }
+        }
+//        System.out.println("URL Params:"+urlParameters); 
+
+        try {
+            URL url = new URL(_httpUrl); 
+            HttpURLConnection connection = (HttpURLConnection) url.openConnection();           
+            connection.setDoOutput(true);
+            connection.setDoInput(true);
+            connection.setInstanceFollowRedirects(true); 
+            connection.setRequestMethod("POST"); 
+            connection.setUseCaches (true);
+            connection.setRequestProperty("Content-length", String.valueOf(urlParameters.length())); 
+            connection.setRequestProperty("Content-Type","application/x-www-form-urlencoded"); 
+            // connection.setRequestProperty("User-Agent", "Mozilla/4.0 (compatible; MSIE 5.0;Windows98;DigExt)"); 
+
+            DataOutputStream outstream = new DataOutputStream(connection.getOutputStream ());
+            outstream.writeBytes(urlParameters);
+            outstream.flush();
+            outstream.close();
+
+            int respCode = connection.getResponseCode();
+            if (respCode != 200)
+            {
+                System.err.println("Response Code:" + respCode); 
+                System.out.println("Response Message:" + connection.getResponseMessage()); 
+//            DataInputStream input = new DataInputStream(connection.getInputStream() ); 
+//            for( int c = input.read(); c != -1; c = input.read() ) 
+//                System.out.print( (char)c ); 
+//                input.close(); 
+            }
+            connection.disconnect();
+        } catch (IOException e) {
+            System.err.println("HttpStatusReporter: Failed to open connection for " + _httpUrl);
+            e.printStackTrace();
+        }
+    }
+
+    public void close() {
+        return;
+    }
+}