Merge pull request #181 from mpoindexter/master

Fix and test for issue #180

h2/src/main/org/h2/table/Table.java

@@ -463,7 +463,16 @@ public abstract class Table extends SchemaObjectBase {
             }
             Row o = rows.next();
             rows.next();
-            removeRow(session, o);
+            try {
+                removeRow(session, o);
+            } catch (DbException e) {
+                if (e.getErrorCode() == ErrorCode.CONCURRENT_UPDATE_1) {
+                    session.rollbackTo(rollback, false);
+                    session.startStatementWithinTransaction();
+                    rollback = session.setSavepoint();
+                }
+                throw e;
+            }
             session.log(this, UndoLogRecord.DELETE, o);
         }
         // add the new rows

h2/src/test/org/h2/test/TestAll.java

@@ -107,6 +107,7 @@ import org.h2.test.jdbcx.TestXASimple;
 import org.h2.test.mvcc.TestMvcc1;
 import org.h2.test.mvcc.TestMvcc2;
 import org.h2.test.mvcc.TestMvcc3;
+import org.h2.test.mvcc.TestMvcc4;
 import org.h2.test.mvcc.TestMvccMultiThreaded;
 import org.h2.test.poweroff.TestReorderWrites;
 import org.h2.test.rowlock.TestRowLocks;
@@ -330,6 +331,11 @@ java org.h2.test.TestAll timer
      */
     public boolean fast;
 
+    /**
+     * The lock timeout to use
+     */
+    public int lockTimeout = 50;
+
     /**
      * If the transaction log should be kept small (that is, the log should be
      * switched early).
@@ -739,6 +745,7 @@ kill -9 `jps -l | grep "org.h2.test." | cut -d " " -f 1`
         addTest(new TestMvcc1());
         addTest(new TestMvcc2());
         addTest(new TestMvcc3());
+        addTest(new TestMvcc4());
         addTest(new TestMvccMultiThreaded());
         addTest(new TestRowLocks());
 

h2/src/test/org/h2/test/TestBase.java

@@ -299,7 +299,7 @@ public abstract class TestBase {
         } else if (config.throttle > 0) {
             url = addOption(url, "THROTTLE", "" + config.throttle);
         }
-        url = addOption(url, "LOCK_TIMEOUT", "50");
+        url = addOption(url, "LOCK_TIMEOUT", "" + config.lockTimeout);
         if (config.diskUndo && admin) {
             url = addOption(url, "MAX_MEMORY_UNDO", "3");
         }

h2/src/test/org/h2/test/mvcc/TestMvcc4.java

+/*
+ * Copyright 2004-2014 H2 Group. Multiple-Licensed under the MPL 2.0,
+ * and the EPL 1.0 (http://h2database.com/html/license.html).
+ * Initial Developer: H2 Group
+ */
+package org.h2.test.mvcc;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+import org.h2.test.TestBase;
+
+/**
+ * Additional MVCC (multi version concurrency) test cases.
+ */
+public class TestMvcc4 extends TestBase {
+
+    /**
+     * Run just this test.
+     *
+     * @param a ignored
+     */
+    public static void main(String... a) throws Exception {
+        TestBase test = TestBase.createCaller().init();
+        test.config.mvcc = true;
+        test.config.lockTimeout = 20000;
+        test.config.memory = true;
+        test.test();
+    }
+
+    @Override
+    public void test() throws SQLException {
+        testSelectForUpdateAndUpdateConcurrency();
+    }
+
+    private void testSelectForUpdateAndUpdateConcurrency() throws SQLException {
+        Connection setup = getConnection("mvcc4");
+        setup.setAutoCommit(false);
+
+        {
+            Statement s = setup.createStatement();
+            s.executeUpdate("CREATE TABLE test ("
+                    + "entity_id VARCHAR(100) NOT NULL PRIMARY KEY, "
+                    + "lastUpdated TIMESTAMP NOT NULL)");
+
+            PreparedStatement ps = setup.prepareStatement("INSERT INTO test (entity_id, lastUpdated) VALUES (?, ?)");
+            for (int i = 0; i < 2; i++) {
+                String id = "" + i;
+                ps.setString(1, id);
+                ps.setTimestamp(2, new Timestamp(System.currentTimeMillis()));
+                ps.executeUpdate();
+            }
+            setup.commit();
+        }
+
+        //Create a connection from thread 1
+        Connection c1 = getConnection("mvcc4");
+        c1.setAutoCommit(false);
+
+        //Fire off a concurrent update.
+        final Thread mainThread = Thread.currentThread();
+        final CountDownLatch executedUpdate = new CountDownLatch(1);
+        new Thread() {
+            public void run() {
+                try {
+                    Connection c2 = getConnection("mvcc4");
+                    c2.setAutoCommit(false);
+
+                    PreparedStatement ps = c2.prepareStatement("SELECT * FROM test WHERE entity_id = ? FOR UPDATE");
+                    ps.setString(1, "1");
+                    ps.executeQuery().next();
+
+                    executedUpdate.countDown();
+                    waitForThreadToBlockOnDB(mainThread);
+
+                    c2.commit();
+                    c2.close();
+                } catch (SQLException e) {
+                    e.printStackTrace();
+                }
+            }
+        }.start();
+
+        //Wait until the concurrent update has executed, but not yet committed
+        try {
+            executedUpdate.await();
+        } catch (InterruptedException e) {
+        }
+
+        {
+            //Execute an update.  This should initially fail, and enter the waiting for lock case.
+            PreparedStatement ps = c1.prepareStatement("UPDATE test SET lastUpdated = ?");
+            ps.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
+            ps.executeUpdate();
+        }
+
+        c1.commit();
+        c1.close();
+
+        Connection verify = getConnection("mvcc4");
+        {
+            verify.setAutoCommit(false);
+            PreparedStatement ps = verify.prepareStatement("SELECT COUNT(*) FROM test");
+            ResultSet rs = ps.executeQuery();
+            assertTrue(rs.next());
+            assertTrue(rs.getInt(1) == 2);
+            verify.commit();
+            verify.close();
+        }
+
+        setup.close();
+    }
+
+    private static void waitForThreadToBlockOnDB(Thread t) {
+        while (true) {
+            Map<Thread, StackTraceElement[]> threadMap = Thread.getAllStackTraces();
+            StackTraceElement[] elements = threadMap.get(t);
+            if (elements != null
+                    && elements.length > 1
+                    && "wait".equals(elements[0].getMethodName())
+                    && "filterConcurrentUpdate".equals(elements[1].getMethodName())) {
+                return;
+            }
+            try {
+                Thread.sleep(10);
+            } catch (InterruptedException e1) {
+            }
+        }
+    }
+}
+
+
+
+