YCSB+T: Added transactions to DB and DBWrapper and validate() to Workloa...

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

@@ -309,57 +309,76 @@ public void run()
     //NOTE: Switching to using nanoTime and parkNanos for time management here such that the measurements
     // and the client thread have the same view on time.
 
-    //spread the thread operations out so they don't all hit the DB at the same time
-    // GH issue 4 - throws exception if _target>1 because random.nextInt argument must be >0
-    // and the sleep() doesn't make sense for granularities < 1 ms anyway
-    if ((_targetOpsPerMs > 0) && (_targetOpsPerMs <= 1.0))
-    {
-      long randomMinorDelay = Utils.random().nextInt((int) _targetOpsTickNs);
-      sleepUntil(System.nanoTime() + randomMinorDelay);
-    }
-    try
-    {
-      if (_dotransactions)
-      {
-        long startTimeNanos = System.nanoTime();
-
-        while (((_opcount == 0) || (_opsdone < _opcount)) && !_workload.isStopRequested())
-        {
-
-          if (!_workload.doTransaction(_db,_workloadstate))
-          {
-            break;
-          }
-
-          _opsdone++;
-
-          throttleNanos(startTimeNanos);
-        }
-      }
-      else
-      {
-        long startTimeNanos = System.nanoTime();
-
-        while (((_opcount == 0) || (_opsdone < _opcount)) && !_workload.isStopRequested())
-        {
-
-          if (!_workload.doInsert(_db,_workloadstate))
-          {
-            break;
-          }
-
-          _opsdone++;
-
-          throttleNanos(startTimeNanos);
-        }
-      }
-    }
-    catch (Exception e)
-    {
-      e.printStackTrace();
-      e.printStackTrace(System.out);
-      System.exit(0);
-    }
+		//spread the thread operations out so they don't all hit the DB at the same time
+		try
+		{
+		   //GH issue 4 - throws exception if _target>1 because random.nextInt argument must be >0
+		   //and the sleep() doesn't make sense for granularities < 1 ms anyway
+		   if ( (_targetOpsPerMs>0) && (_targetOpsPerMs<=1.0) ) 
+		   {
+		      sleep(Utils.random().nextInt((int)(1.0/_targetOpsTickNs)));
+		   }
+		}
+		catch (InterruptedException e)
+		{
+		  // do nothing.
+		}
+
+		try
+		{
+			if (_dotransactions)
+			{
+                                long startTimeNanos = System.nanoTime();
+
+				while (((_opcount == 0) || (_opsdone < _opcount)) && !_workload.isStopRequested())
+				{
+
+					try {
+						_db.start();
+						if (_workload.doTransaction(_db,_workloadstate)) {
+							_db.commit();
+						} else {
+							_db.abort();							
+						}
+					} catch (DBException e) {
+						throw new WorkloadException(e);
+					}
+
+					_opsdone++;
+
+                                        throttleNanos(startTimeNanos);
+				}
+			}
+			else
+			{
+                                long startTimeNanos = System.nanoTime();
+
+				while (((_opcount == 0) || (_opsdone < _opcount)) && !_workload.isStopRequested())
+				{
+
+					try {
+						_db.start();
+						if (_workload.doInsert(_db,_workloadstate)) {
+							_db.commit();
+						} else {
+							_db.abort();
+						}
+					} catch (DBException e) {
+						throw new WorkloadException(e);
+					}
+
+					_opsdone++;
+
+                                        throttleNanos(startTimeNanos);
+				}
+			}
+		}
+		catch (Exception e)
+		{
+			e.printStackTrace();
+			e.printStackTrace(System.out);
+			System.exit(0);
+		}
 
     try
     {
@@ -930,16 +949,32 @@ public void run()
       }
     }
 
-    try
-    {
-      workload.cleanup();
-    }
-    catch (WorkloadException e)
-    {
-      e.printStackTrace();
-      e.printStackTrace(System.out);
-      System.exit(0);
-    }
+		try {
+			DB db = DBFactory.newDB(dbname,props);
+			db.init();
+			if (workload.validate(db))
+				System.out.println("Database validation succeeded");
+			else
+				System.out.println("Database validation failed");
+		} catch (WorkloadException e) {
+			System.out.println("Database validation failed with error: " + e.getMessage());
+		} catch (UnknownDBException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		} catch (DBException e) {
+			e.printStackTrace();
+		}
+
+		try
+		{
+			workload.cleanup();
+		}
+		catch (WorkloadException e)
+		{
+			e.printStackTrace();
+			e.printStackTrace(System.out);
+			System.exit(0);
+		}
 
     try
     {

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

@@ -81,6 +81,27 @@ public void cleanup() throws DBException
 	{
 	}
 
+	/**
+	 * Start a database transaction. 
+	 */
+	public void start() throws DBException
+	{
+	}
+
+	/**
+	 * Commit the current database transaction. 
+	 */
+	public void commit() throws DBException
+	{
+	}
+
+	/**
+	 * Abort the current database transaction. 
+	 */
+	public void abort() throws DBException
+	{
+	}
+
 	/**
 	 * Read a record from the database. Each field/value pair from the result will be stored in a HashMap.
 	 *

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

@@ -107,6 +107,52 @@ public void cleanup() throws DBException
     measure("CLEANUP", Status.OK, ist, st, en);
   }
 
+	public void start() throws DBException
+	{
+		long st=System.nanoTime();
+		try {
+			_db.start();
+			long en = System.nanoTime();
+			_measurements.measure("START", (int) ((en - st) / 1000));
+			_measurements.reportStatus("START", Status.OK);
+		} catch (DBException e) {
+			long en=System.nanoTime();
+			_measurements.measure("START",(int)((en-st)/1000));
+			_measurements.reportStatus("START", Status.ERROR);
+			throw e;
+		}
+	}
+
+	public void commit() throws DBException
+	{
+		long st=System.nanoTime();
+		try {
+			_db.commit();
+			long en = System.nanoTime();
+			_measurements.measure("COMMIT", (int) ((en - st) / 1000));
+			_measurements.reportStatus("COMMIT", Status.OK);
+		} catch (DBException e) {
+			long en = System.nanoTime();
+			_measurements.measure("ABORT", (int) ((en - st) / 1000));
+			_measurements.reportStatus("ABORT", Status.ERROR);
+		}
+	}
+
+	public void abort() throws DBException
+	{
+		long st=System.nanoTime();
+		try {
+			_db.abort();
+			long en = System.nanoTime();
+			_measurements.measure("ABORT", (int) ((en - st) / 1000));
+			_measurements.reportStatus("ABORT", Status.OK);
+		} catch (DBException e) {
+			long en=System.nanoTime();
+			_measurements.measure("ABORT",(int)((en-st)/1000));
+			_measurements.reportStatus("ABORT", Status.ERROR);
+		}
+	}
+
   /**
    * Read a record from the database. Each field/value pair from the result
    * will be stored in a HashMap.

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

@@ -80,8 +80,9 @@ public void cleanup() throws WorkloadException
        * other, and it will be difficult to reach the target throughput. Ideally, this function would have no side
        * effects other than DB operations and mutations on threadstate. Mutations to threadstate do not need to be
        * synchronized, since each thread has its own threadstate instance.
+       * @throws WorkloadException 
        */
-      public abstract boolean doInsert(DB db, Object threadstate);
+      public abstract boolean doInsert(DB db, Object threadstate) throws WorkloadException;
 
       /**
        * Do one transaction operation. Because it will be called concurrently from multiple client threads, this 
@@ -91,9 +92,11 @@ public void cleanup() throws WorkloadException
        * synchronized, since each thread has its own threadstate instance.
        * 
        * @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.
+       * @throws WorkloadException 
        */
-      public abstract boolean doTransaction(DB db, Object threadstate);
-
+      public abstract boolean doTransaction(DB db, Object threadstate) throws WorkloadException;
+
+
       /**
        * Allows scheduling a request to stop the workload.
        */
@@ -109,4 +112,14 @@ public boolean isStopRequested() {
         if (stopRequested.get() == true) return true;
         else return false;
       }
+
+     /**
+      * Perform validation of the database db after the workload has executed.
+      * 
+      * @return false if the workload left the database in an inconsistent state, true if it is consistent.
+      * @throws WorkloadException 
+      */
+     public boolean validate(DB db) throws WorkloadException {
+    	 return true;
+     }
 }