Merge branch 'master' of git://github.com/sudiptodas/YCSB into sudiptodas-master

src/com/yahoo/ycsb/Client.java

@@ -220,7 +220,7 @@ class ClientThread extends Thread
 		}
 		catch (InterruptedException e)
 		{
-		   //do nothing
+		  // do nothing.
 		}
 		
 		try
@@ -229,7 +229,7 @@ class ClientThread extends Thread
 			{
 				long st=System.currentTimeMillis();
 
-				while ( (_opcount==0) || (_opsdone<_opcount) )
+				while (((_opcount == 0) || (_opsdone < _opcount)) && !_workload.isStopRequested())
 				{
 
 					if (!_workload.doTransaction(_db,_workloadstate))
@@ -254,7 +254,7 @@ class ClientThread extends Thread
 							}
 							catch (InterruptedException e)
 							{
-								//do nothing
+							  // do nothing.
 							}
 
 						}
@@ -265,7 +265,7 @@ class ClientThread extends Thread
 			{
 				long st=System.currentTimeMillis();
 
-				while ( (_opcount==0) || (_opsdone<_opcount) )
+				while (((_opcount == 0) || (_opsdone < _opcount)) && !_workload.isStopRequested())
 				{
 
 					if (!_workload.doInsert(_db,_workloadstate))
@@ -290,7 +290,7 @@ class ClientThread extends Thread
 							}
 							catch (InterruptedException e)
 							{
-								//do nothing
+							  // do nothing.
 							}
 						}
 					}
@@ -335,6 +335,11 @@ public class Client
 	 * should support the "insertstart" property, which tells them which record to start at.
 	 */
 	public static final String INSERT_COUNT_PROPERTY="insertcount";
+	
+	/**
+   * The maximum amount of time (in seconds) for which the benchmark will be run.
+   */
+  public static final String MAX_EXECUTION_TIME = "maxexecutiontime";
 
 	public static void usageMessage()
 	{
@@ -599,6 +604,8 @@ public class Client
 		{
 			System.exit(0);
 		}
+		
+		long maxExecutionTime = Integer.parseInt(props.getProperty(MAX_EXECUTION_TIME, "0"));
 
 		//get number of threads, target and db
 		threadcount=Integer.parseInt(props.getProperty("threadcount","1"));
@@ -738,12 +745,22 @@ public class Client
 		{
 			t.start();
 		}
+		
+    Thread terminator = null;
+    
+    if (maxExecutionTime > 0) {
+      terminator = new TerminatorThread(maxExecutionTime, threads, workload);
+      terminator.start();
+    }
+    
+    int opsDone = 0;
 
 		for (Thread t : threads)
 		{
 			try
 			{
 				t.join();
+				opsDone += ((ClientThread)t).getOpsDone();
 			}
 			catch (InterruptedException e)
 			{
@@ -751,6 +768,10 @@ public class Client
 		}
 
 		long en=System.currentTimeMillis();
+		
+		if (terminator != null && !terminator.isInterrupted()) {
+      terminator.interrupt();
+    }
 
 		if (status)
 		{
@@ -770,7 +791,7 @@ public class Client
 
 		try
 		{
-			exportMeasurements(props, opcount, en - st);
+			exportMeasurements(props, opsDone, en - st);
 		} catch (IOException e)
 		{
 			System.err.println("Could not export measurements, error: " + e.getMessage());

src/com/yahoo/ycsb/TerminatorThread.java

+/**
+ * Copyright (c) 2011 Yahoo! Inc. All rights reserved.                                                                                                                             
+ *                                                                                                                                                                                 
+ * 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. See accompanying                                                                                                                 
+ * LICENSE file.
+ */
+package com.yahoo.ycsb;
+
+import java.util.Vector;
+
+/**
+ * A thread that waits for the maximum specified time and then interrupts all the client
+ * threads passed as the Vector at initialization of this thread.
+ * 
+ * The maximum execution time passed is assumed to be in seconds.
+ * 
+ * @author sudipto
+ *
+ */
+public class TerminatorThread extends Thread {
+  
+  private Vector<Thread> threads;
+  private long maxExecutionTime;
+  private Workload workload;
+  private long waitTimeOutInMS;
+  
+  public TerminatorThread(long maxExecutionTime, Vector<Thread> threads, 
+      Workload workload) {
+    this.maxExecutionTime = maxExecutionTime;
+    this.threads = threads;
+    this.workload = workload;
+    waitTimeOutInMS = 2000;
+    System.err.println("Maximum execution time specified as: " + maxExecutionTime + " secs");
+  }
+  
+  public void run() {
+    try {
+      Thread.sleep(maxExecutionTime * 1000);
+    } catch (InterruptedException e) {
+      System.err.println("Could not wait until max specified time, TerminatorThread interrupted.");
+      return;
+    }
+    System.err.println("Maximum time elapsed. Requesting stop for the workload.");
+    workload.requestStop();
+    System.err.println("Stop requested for workload. Now Joining!");
+    for (Thread t : threads) {
+      while (t.isAlive()) {
+        try {
+          t.join(waitTimeOutInMS);
+          if (t.isAlive()) {
+            System.err.println("Still waiting for thread " + t.getName() + " to complete. " +
+                "Workload status: " + workload.isStopRequested());
+          }
+        } catch (InterruptedException e) {
+          // Do nothing. Don't know why I was interrupted.
+        }
+      }
+    }
+  }
+}

src/com/yahoo/ycsb/Workload.java

@@ -18,6 +18,7 @@
 package com.yahoo.ycsb;
 
 import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * One experiment scenario. One object of this type will
@@ -39,6 +40,8 @@ public abstract class Workload
 	
 	public static final String INSERT_START_PROPERTY_DEFAULT="0";
 	
+	private volatile AtomicBoolean stopRequested = new AtomicBoolean(false);
+	
       /**
        * Initialize the scenario. Create any generators and other shared objects here.
        * Called once, in the main client thread, before any operations are started.
@@ -90,4 +93,20 @@ public abstract class Workload
        * @return false if the workload knows it is done for this thread. Client will terminate the thread. Return true otherwise. Return true for workloads that rely on operationcount. For workloads that read traces from a file, return true when there are more to do, false when you are done.
        */
       public abstract boolean doTransaction(DB db, Object threadstate);
+      
+      /**
+       * Allows scheduling a request to stop the workload.
+       */
+      public void requestStop() {
+        stopRequested.set(true);
+      }
+      
+      /**
+       * Check the status of the stop request flag.
+       * @return true if stop was requested, false otherwise.
+       */
+      public boolean isStopRequested() {
+        if (stopRequested.get() == true) return true;
+        else return false;
+      }
 }