提交 26f915bd authored 作者: Thomas Mueller's avatar Thomas Mueller

Fix for issue #143, deadlock between two sessions hitting the same sequence on a column.

上级 ebf435d5
...@@ -983,15 +983,17 @@ public class Database implements DataHandler { ...@@ -983,15 +983,17 @@ public class Database implements DataHandler {
* @param session the session * @param session the session
* @param obj the object to add * @param obj the object to add
*/ */
public synchronized void addSchemaObject(Session session, SchemaObject obj) { public void addSchemaObject(Session session, SchemaObject obj) {
int id = obj.getId(); int id = obj.getId();
if (id > 0 && !starting) { if (id > 0 && !starting) {
checkWritingAllowed(); checkWritingAllowed();
} }
lockMeta(session); lockMeta(session);
synchronized (this) {
obj.getSchema().add(obj); obj.getSchema().add(obj);
addMeta(session, obj); addMeta(session, obj);
} }
}
/** /**
* Add an object to the database. * Add an object to the database.
...@@ -1791,7 +1793,7 @@ public class Database implements DataHandler { ...@@ -1791,7 +1793,7 @@ public class Database implements DataHandler {
* @param session the session * @param session the session
* @param obj the object to be removed * @param obj the object to be removed
*/ */
public synchronized void removeSchemaObject(Session session, public void removeSchemaObject(Session session,
SchemaObject obj) { SchemaObject obj) {
int type = obj.getType(); int type = obj.getType();
if (type == DbObject.TABLE_OR_VIEW) { if (type == DbObject.TABLE_OR_VIEW) {
...@@ -1817,6 +1819,7 @@ public class Database implements DataHandler { ...@@ -1817,6 +1819,7 @@ public class Database implements DataHandler {
} }
checkWritingAllowed(); checkWritingAllowed();
lockMeta(session); lockMeta(session);
synchronized (this) {
Comment comment = findComment(obj); Comment comment = findComment(obj);
if (comment != null) { if (comment != null) {
removeDatabaseObject(session, comment); removeDatabaseObject(session, comment);
...@@ -1834,6 +1837,7 @@ public class Database implements DataHandler { ...@@ -1834,6 +1837,7 @@ public class Database implements DataHandler {
} }
removeMeta(session, id); removeMeta(session, id);
} }
}
/** /**
* Check if this database disk-based. * Check if this database disk-based.
......
...@@ -33,12 +33,8 @@ public class Sequence extends SchemaObjectBase { ...@@ -33,12 +33,8 @@ public class Sequence extends SchemaObjectBase {
private long maxValue; private long maxValue;
private boolean cycle; private boolean cycle;
private boolean belongsToTable; private boolean belongsToTable;
private Object flushSync = new Object();
/** private boolean writeWithMargin;
* The last valueWithMargin we flushed. We do a little dance with this to
* avoid an ABBA deadlock.
*/
private long lastFlushValueWithMargin;
/** /**
* Creates a new sequence for an auto-increment column. * Creates a new sequence for an auto-increment column.
...@@ -217,15 +213,16 @@ public class Sequence extends SchemaObjectBase { ...@@ -217,15 +213,16 @@ public class Sequence extends SchemaObjectBase {
@Override @Override
public synchronized String getCreateSQL() { public synchronized String getCreateSQL() {
long v = writeWithMargin ? valueWithMargin : value;
StringBuilder buff = new StringBuilder("CREATE SEQUENCE "); StringBuilder buff = new StringBuilder("CREATE SEQUENCE ");
buff.append(getSQL()).append(" START WITH ").append(value); buff.append(getSQL()).append(" START WITH ").append(v);
if (increment != 1) { if (increment != 1) {
buff.append(" INCREMENT BY ").append(increment); buff.append(" INCREMENT BY ").append(increment);
} }
if (minValue != getDefaultMinValue(value, increment)) { if (minValue != getDefaultMinValue(v, increment)) {
buff.append(" MINVALUE ").append(minValue); buff.append(" MINVALUE ").append(minValue);
} }
if (maxValue != getDefaultMaxValue(value, increment)) { if (maxValue != getDefaultMaxValue(v, increment)) {
buff.append(" MAXVALUE ").append(maxValue); buff.append(" MAXVALUE ").append(maxValue);
} }
if (cycle) { if (cycle) {
...@@ -248,13 +245,11 @@ public class Sequence extends SchemaObjectBase { ...@@ -248,13 +245,11 @@ public class Sequence extends SchemaObjectBase {
*/ */
public long getNext(Session session) { public long getNext(Session session) {
boolean needsFlush = false; boolean needsFlush = false;
long retVal; long result;
long flushValueWithMargin = -1;
synchronized (this) { synchronized (this) {
if ((increment > 0 && value >= valueWithMargin) || if ((increment > 0 && value >= valueWithMargin) ||
(increment < 0 && value <= valueWithMargin)) { (increment < 0 && value <= valueWithMargin)) {
valueWithMargin += increment * cacheSize; valueWithMargin += increment * cacheSize;
flushValueWithMargin = valueWithMargin;
needsFlush = true; needsFlush = true;
} }
if ((increment > 0 && value > maxValue) || if ((increment > 0 && value > maxValue) ||
...@@ -262,19 +257,18 @@ public class Sequence extends SchemaObjectBase { ...@@ -262,19 +257,18 @@ public class Sequence extends SchemaObjectBase {
if (cycle) { if (cycle) {
value = increment > 0 ? minValue : maxValue; value = increment > 0 ? minValue : maxValue;
valueWithMargin = value + (increment * cacheSize); valueWithMargin = value + (increment * cacheSize);
flushValueWithMargin = valueWithMargin;
needsFlush = true; needsFlush = true;
} else { } else {
throw DbException.get(ErrorCode.SEQUENCE_EXHAUSTED, getName()); throw DbException.get(ErrorCode.SEQUENCE_EXHAUSTED, getName());
} }
} }
retVal = value; result = value;
value += increment; value += increment;
} }
if (needsFlush) { if (needsFlush) {
flush(session, flushValueWithMargin); flush(session);
} }
return retVal; return result;
} }
/** /**
...@@ -283,7 +277,7 @@ public class Sequence extends SchemaObjectBase { ...@@ -283,7 +277,7 @@ public class Sequence extends SchemaObjectBase {
public void flushWithoutMargin() { public void flushWithoutMargin() {
if (valueWithMargin != value) { if (valueWithMargin != value) {
valueWithMargin = value; valueWithMargin = value;
flush(null, valueWithMargin); flush(null);
} }
} }
...@@ -291,47 +285,39 @@ public class Sequence extends SchemaObjectBase { ...@@ -291,47 +285,39 @@ public class Sequence extends SchemaObjectBase {
* Flush the current value, including the margin, to disk. * Flush the current value, including the margin, to disk.
* *
* @param session the session * @param session the session
* @param flushValueWithMargin whether to reserve more entries
*/ */
public void flush(Session session, long flushValueWithMargin) { public void flush(Session session) {
if (isTemporary()) {
return;
}
if (session == null || !database.isSysTableLockedBy(session)) { if (session == null || !database.isSysTableLockedBy(session)) {
// This session may not lock the sys table (except if it already has // This session may not lock the sys table (except if it already has
// locked it) because it must be committed immediately, otherwise // locked it) because it must be committed immediately, otherwise
// other threads can not access the sys table. // other threads can not access the sys table.
Session sysSession = database.getSystemSession(); Session sysSession = database.getSystemSession();
synchronized (sysSession) { synchronized (sysSession) {
flushInternal(sysSession, flushValueWithMargin); synchronized (flushSync) {
flushInternal(sysSession);
}
sysSession.commit(false); sysSession.commit(false);
} }
} else { } else {
synchronized (session) { synchronized (session) {
flushInternal(session, flushValueWithMargin); synchronized (flushSync) {
flushInternal(session);
}
} }
} }
} }
private void flushInternal(Session session, long flushValueWithMargin) { private void flushInternal(Session session) {
final boolean metaWasLocked = database.lockMeta(session); final boolean metaWasLocked = database.lockMeta(session);
synchronized (this) { // just for this case, use the value with the margin
if (flushValueWithMargin == lastFlushValueWithMargin) {
if (!metaWasLocked) {
database.unlockMeta(session);
}
return;
}
}
// just for this case, use the value with the margin for the script
long realValue = value;
try { try {
value = valueWithMargin; writeWithMargin = true;
if (!isTemporary()) {
database.updateMeta(session, this); database.updateMeta(session, this);
}
} finally { } finally {
value = realValue; writeWithMargin = false;
}
synchronized (this) {
lastFlushValueWithMargin = flushValueWithMargin;
} }
if (!metaWasLocked) { if (!metaWasLocked) {
database.unlockMeta(session); database.unlockMeta(session);
......
...@@ -348,7 +348,7 @@ public class Column { ...@@ -348,7 +348,7 @@ public class Column {
if (update) { if (update) {
sequence.modify(now + inc, null, null, null); sequence.modify(now + inc, null, null, null);
session.setLastIdentity(ValueLong.get(now)); session.setLastIdentity(ValueLong.get(now));
sequence.flush(session, 0); sequence.flush(session);
} }
} }
} }
......
...@@ -13,6 +13,8 @@ import java.sql.Statement; ...@@ -13,6 +13,8 @@ import java.sql.Statement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import org.h2.api.Trigger;
import org.h2.test.TestBase; import org.h2.test.TestBase;
import org.h2.util.Task; import org.h2.util.Task;
...@@ -33,7 +35,6 @@ public class TestSequence extends TestBase { ...@@ -33,7 +35,6 @@ public class TestSequence extends TestBase {
@Override @Override
public void test() throws Exception { public void test() throws Exception {
testConcurrentCreate(); testConcurrentCreate();
;if(true)return;
testSchemaSearchPath(); testSchemaSearchPath();
testAlterSequenceColumn(); testAlterSequenceColumn();
testAlterSequence(); testAlterSequence();
...@@ -49,24 +50,17 @@ public class TestSequence extends TestBase { ...@@ -49,24 +50,17 @@ public class TestSequence extends TestBase {
} }
private void testConcurrentCreate() throws Exception { private void testConcurrentCreate() throws Exception {
while(true)
try {
testConcurrentCreate2();
} catch(Exception e) {
System.out.println(e);
}
}
private void testConcurrentCreate2() throws Exception {
deleteDb("sequence"); deleteDb("sequence");
final String url = getURL("sequence;MULTI_THREADED=1", true); final String url = getURL("sequence;MULTI_THREADED=1;LOCK_TIMEOUT=2000", true);
Connection conn = getConnection(url); Connection conn = getConnection(url);
Task[] tasks = new Task[2]; Task[] tasks = new Task[2];
try { try {
Statement stat = conn.createStatement(); Statement stat = conn.createStatement();
stat.execute("create table dummy(id bigint primary key)");
stat.execute("create table test(id bigint primary key)"); stat.execute("create table test(id bigint primary key)");
stat.execute("create sequence test_seq cache 2"); stat.execute("create sequence test_seq cache 2");
for (int i = 0; i < tasks.length; i++) { for (int i = 0; i < tasks.length; i++) {
final int x = i;
tasks[i] = new Task() { tasks[i] = new Task() {
@Override @Override
public void call() throws Exception { public void call() throws Exception {
...@@ -81,18 +75,30 @@ public class TestSequence extends TestBase { ...@@ -81,18 +75,30 @@ public class TestSequence extends TestBase {
if (Math.random() < 0.01) { if (Math.random() < 0.01) {
prep2.execute(); prep2.execute();
} }
if (Math.random() < 0.01) {
createDropTrigger(conn);
}
} }
} finally { } finally {
conn.close(); conn.close();
} }
} }
private void createDropTrigger(Connection conn) throws Exception {
String triggerName = "t_" + x;
Statement stat = conn.createStatement();
stat.execute("create trigger " + triggerName +
" before insert on dummy call \"" +
TriggerTest.class.getName() + "\"");
stat.execute("drop trigger " + triggerName);
}
}.execute(); }.execute();
} }
Thread.sleep(100); Thread.sleep(1000);
for (Task t : tasks) { for (Task t : tasks) {
t.get(); t.get();
} }
stat.execute("shutdown immediately");
} finally { } finally {
for (Task t : tasks) { for (Task t : tasks) {
t.join(); t.join();
...@@ -413,4 +419,35 @@ public class TestSequence extends TestBase { ...@@ -413,4 +419,35 @@ public class TestSequence extends TestBase {
long value = rs.getLong(1); long value = rs.getLong(1);
return value; return value;
} }
/**
* A test trigger.
*/
public static class TriggerTest implements Trigger {
@Override
public void init(Connection conn, String schemaName,
String triggerName, String tableName, boolean before, int type)
throws SQLException {
conn.createStatement().executeQuery("call next value for test_seq");
}
@Override
public void fire(Connection conn, Object[] oldRow, Object[] newRow)
throws SQLException {
// ignore
}
@Override
public void close() throws SQLException {
// ignore
}
@Override
public void remove() throws SQLException {
// ignore
}
}
} }
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论